class PlatformService extends PlatformServiceGrpc.PlatformServiceImplBase implements AxonServerClientService {

    @Override
    public StreamObserver<PlatformInboundInstruction> openStream(
            StreamObserver<PlatformOutboundInstruction> responseObserver) {
        String context = contextProvider.getContext();
        SendingStreamObserver<PlatformOutboundInstruction> sendingStreamObserver =
                new SendingStreamObserver<>(responseObserver);

        return new ReceivingStreamObserver<PlatformInboundInstruction>(logger) {
            private final AtomicReference<ClientComponent> clientComponent = new AtomicReference<>();

            @Override
            protected void consume(PlatformInboundInstruction instruction) {
                RequestCase requestCase = instruction.getRequestCase();
                if (instruction.hasRegister()) { // TODO: 11/1/2019 register this as instruction handler
                    instructionAckSource.sendSuccessfulAck(instruction.getInstructionId(), sendingStreamObserver);
                    ClientIdentification client = instruction.getRegister();
                    String clientId = client.getClientId();
                    String clientStreamId = clientId + "." + UUID.randomUUID().toString();
                    clientIdRegistry.register(clientStreamId, clientId, ClientIdRegistry.ConnectionType.PLATFORM);
                    eventPublisher.publishEvent(new ClientTagsUpdate(clientStreamId,
                                                                     context,
                                                                     client.getTagsMap()));

                    ClientComponent clientComponent = new ClientComponent(clientStreamId,
                                                                          clientId,
                                                                          client.getComponentName(),
                                                                          context);
                    this.clientComponent.compareAndSet(null, clientComponent);
                    registerClient(this.clientComponent.get(), sendingStreamObserver);
                    eventPublisher.publishEvent(new ClientVersionUpdate(clientStreamId,
                                                                        context,
                                                                        client.getVersion()));
                } else if (!handlers.containsKey(requestCase)) {
                    instructionAckSource.sendUnsupportedInstruction(instruction.getInstructionId(),
                                                                    topology.getMe().getName(),
                                                                    sendingStreamObserver);
                } else {
                    handlers.getOrDefault(requestCase, new ArrayDeque<>())
                            .forEach(consumer -> {
                                instructionAckSource.sendSuccessfulAck(instruction.getInstructionId(),
                                                                       sendingStreamObserver);
                                consumer.accept(this.clientComponent.get(),
                                                instruction);
                            });
                }
            }

            @Override
            protected String sender() {
                return clientComponent.get() == null ? null : clientComponent.get().clientStreamId;
            }

            @Override
            public void onError(Throwable throwable) {
                if (!ExceptionUtils.isCancelled(throwable)) {
                    logger.warn("{}: error on connection - {}", sender(), throwable.getMessage());
                }
                deregisterClient(clientComponent.get());
            }

            @Override
            public void onCompleted() {
                deregisterClient(clientComponent.get());
            }
        };
    }

    public  PlatformService(Topology topology,
                           ContextProvider contextProvider,
                           ClientIdRegistry clientIdRegistry,
                           ApplicationEventPublisher eventPublisher,
                           @Qualifier("platformInstructionAckSource")
                                   InstructionAckSource<PlatformOutboundInstruction> instructionAckSource);

}