class AbstractConnector<N, L> implements Notifier<N>,
                                                              StreamObserver<PulseResponse> {

        Listener<L> listener;
        StreamObserver<PulseRequest> reqStream;
        PulseType pulseType;
        PulseRequest.Builder reqBuilder = PulseRequest.newBuilder();
        PulseAckRequest.Builder ackBuilder = PulseAckRequest.newBuilder();

        private AbstractConnector(Listener<L> listener, PulseType pulseType) {
            this.listener = listener;
            this.pulseType = pulseType;
            this.init();
        }

        void init() {
            PulseCreateRequest.Builder builder = PulseCreateRequest.newBuilder()
                                                                   .setPulseType(this.pulseType);

            this.reqStream = PDPulseImpl.this.stub.pulse(this);
            this.reqStream.onNext(reqBuilder.clear().setCreateRequest(builder).build());
        }

        /*** notifier ***/
        @Override
        public void close() {
            this.reqStream.onCompleted();
        }

        @Override
        public abstract void notifyServer(N t);

        @Override
        public void crash(String error) {
            this.reqStream.onError(new Throwable(error));
        }

        /*** listener  ***/
        @Override
        public abstract void onNext(PulseResponse pulseResponse);

        @Override
        public void onError(Throwable throwable) {
            this.listener.onError(throwable);
        }

        @Override
        public void onCompleted() {
            this.listener.onCompleted();
        }

        protected void ackNotice(long noticeId, long observerId) {<FILL_FUNCTION_BODY>}
    }

            threadPool.execute(() -> {
                // log.info("send ack: {}, ts: {}", noticeId, System.currentTimeMillis());
                this.reqStream.onNext(reqBuilder.clear()
                                                .setAckRequest(
                                                        this.ackBuilder.clear()
                                                                       .setNoticeId(noticeId)
                                                                       .setObserverId(observerId)
                                                                       .build()
                                                ).build()
                );
            });
        
<no_super_class>