class CommandService implements AxonServerClientService {

    public StreamObserver<CommandProviderOutbound> openStream(
            StreamObserver<SerializedCommandProviderInbound> responseObserver) {
        String context = contextProvider.getContext();
        SendingStreamObserver<SerializedCommandProviderInbound> wrappedResponseObserver = new SendingStreamObserver<>(
                responseObserver);
        return new ReceivingStreamObserver<CommandProviderOutbound>(logger) {
            private final AtomicReference<String> clientIdRef = new AtomicReference<>();
            private final AtomicReference<ClientStreamIdentification> clientRef = new AtomicReference<>();
            private final AtomicReference<GrpcCommandDispatcherListener> listenerRef = new AtomicReference<>();
            private final AtomicReference<CommandHandler<?>> commandHandlerRef = new AtomicReference<>();

            @Override
            protected void consume(CommandProviderOutbound commandFromSubscriber) {
                switch (commandFromSubscriber.getRequestCase()) {
                    case SUBSCRIBE:
                        instructionAckSource.sendSuccessfulAck(commandFromSubscriber.getInstructionId(),
                                                               wrappedResponseObserver);
                        CommandSubscription subscribe = commandFromSubscriber.getSubscribe();

                        checkInitClient(subscribe.getClientId(), subscribe.getComponentName());
                        SubscribeCommand event = new SubscribeCommand(context,
                                                                      clientRef.get().getClientStreamId(),
                                                                      subscribe,
                                                                      commandHandlerRef.get());
                        eventPublisher.publishEvent(event);
                        break;
                    case UNSUBSCRIBE:
                        instructionAckSource.sendSuccessfulAck(commandFromSubscriber.getInstructionId(),
                                                               wrappedResponseObserver);
                        if (clientRef.get() != null) {
                            UnsubscribeCommand unsubscribe =
                                    new UnsubscribeCommand(context,
                                                           clientRef.get().getClientStreamId(),
                                                           commandFromSubscriber.getUnsubscribe(),
                                                           false);
                            eventPublisher.publishEvent(unsubscribe);
                        }
                        break;
                    case FLOW_CONTROL:
                        instructionAckSource.sendSuccessfulAck(commandFromSubscriber.getInstructionId(),
                                                               wrappedResponseObserver);
                        flowControl(commandFromSubscriber.getFlowControl());
                        break;
                    case COMMAND_RESPONSE:
                        instructionAckSource.sendSuccessfulAck(commandFromSubscriber.getInstructionId(),
                                                               wrappedResponseObserver);
                        commandDispatcher.handleResponse(new SerializedCommandResponse(commandFromSubscriber
                                                                                               .getCommandResponse()),
                                                         false);
                        break;
                    case ACK:
                        InstructionAck ack = commandFromSubscriber.getAck();
                        if (isUnsupportedInstructionErrorResult(ack)) {
                            logger.warn("Unsupported command instruction sent to the client of context {}.", context);
                        } else {
                            logger.trace("Received command instruction ack {}.", ack);
                        }
                        break;
                    default:
                        instructionAckSource.sendUnsupportedInstruction(commandFromSubscriber.getInstructionId(),
                                                                        topology.getMe().getName(),
                                                                        wrappedResponseObserver);
                        break;
                }
            }

            private void initClientReference(String clientId) {
                String clientStreamId = clientId + "." + UUID.randomUUID().toString();
                if (clientRef.compareAndSet(null, new ClientStreamIdentification(context, clientStreamId))) {
                    clientIdRegistry.register(clientStreamId, clientId, ClientIdRegistry.ConnectionType.COMMAND);
                }
                clientIdRef.compareAndSet(null, clientId);
            }

            private void flowControl(FlowControl flowControl) {
                initClientReference(flowControl.getClientId());
                if (listenerRef.compareAndSet(null,
                                              new GrpcCommandDispatcherListener(commandDispatcher.getCommandQueues(),
                                                                                clientRef.get().toString(),
                                                                                wrappedResponseObserver,
                                                                                processingThreads))) {
                    dispatcherListeners.put(clientRef.get(), listenerRef.get());
                }
                listenerRef.get().addPermits(flowControl.getPermits());
            }

            private void checkInitClient(String clientId, String component) {
                initClientReference(clientId);
                commandHandlerRef.compareAndSet(null,
                                                new DirectCommandHandler(wrappedResponseObserver,
                                                                         clientRef.get(),
                                                                         clientId,
                                                                         component));
            }

            @Override
            protected String sender() {
                return clientRef.toString();
            }

            @Override
            public void onError(Throwable cause) {
                if (!ExceptionUtils.isCancelled(cause)) {
                    logger.warn("{}: Error on connection from subscriber - {}", clientRef, cause.getMessage());
                }
                cleanup();
            }

            private void cleanup() {
                if (clientRef.get() != null) {
                    String clientStreamId = clientRef.get().getClientStreamId();
                    String clientId = clientIdRef.get();
                    clientIdRegistry.unregister(clientStreamId, ClientIdRegistry.ConnectionType.COMMAND);
                    eventPublisher.publishEvent(new CommandHandlerDisconnected(clientRef.get().getContext(),
                                                                               clientId,
                                                                               clientStreamId));
                }
                GrpcCommandDispatcherListener listener = listenerRef.get();
                if (listener != null) {
                    listener.cancel();
                    dispatcherListeners.remove(clientRef.get());
                }
                StreamObserverUtils.complete(responseObserver);
            }

            @Override
            public void onCompleted() {
                logger.debug("{}: Connection to subscriber closed by subscriber", clientRef);
                cleanup();
            }
        };
    }

    public  CommandService(Topology topology,
                          CommandDispatcher commandDispatcher,
                          ContextProvider contextProvider,
                          ClientIdRegistry clientIdRegistry,
                          ApplicationEventPublisher eventPublisher,
                          @Qualifier("commandInstructionAckSource")
                                  InstructionAckSource<SerializedCommandProviderInbound> instructionAckSource);

}