@Test public void unsupportedInstruction() { FakeStreamObserver<PlatformOutboundInstruction> responseStream = new FakeStreamObserver<>(); StreamObserver<PlatformInboundInstruction> requestStream = platformService.openStream(responseStream); String instructionId = "instructionId"; requestStream.onNext(PlatformInboundInstruction.newBuilder() .setInstructionId(instructionId) .build()); InstructionAck ack = responseStream.values().get(responseStream.values().size() - 1).getAck(); assertEquals(instructionId, ack.getInstructionId()); assertTrue(ack.hasError()); assertEquals(ErrorCode.UNSUPPORTED_INSTRUCTION.getCode(), ack.getError().getErrorCode()); }
@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()) { 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()); } }; }
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()) { 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()); } }; } }
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()) { 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()); } }; } PlatformService(Topology topology,
                           ContextProvider contextProvider,
                           ClientIdRegistry clientIdRegistry,
                           ApplicationEventPublisher eventPublisher,
                           @Qualifier("platformInstructionAckSource")
                                   InstructionAckSource<PlatformOutboundInstruction> instructionAckSource); }
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()) { 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()); } }; } PlatformService(Topology topology,
                           ContextProvider contextProvider,
                           ClientIdRegistry clientIdRegistry,
                           ApplicationEventPublisher eventPublisher,
                           @Qualifier("platformInstructionAckSource")
                                   InstructionAckSource<PlatformOutboundInstruction> instructionAckSource); @Override void getPlatformServer(ClientIdentification request, StreamObserver<PlatformInfo> responseObserver); @Override StreamObserver<PlatformInboundInstruction> openStream(
            StreamObserver<PlatformOutboundInstruction> responseObserver); boolean requestReconnect(ClientComponent clientName); boolean requestReconnect(String clientId); void sendToAllClients(PlatformOutboundInstruction instruction); void sendToClientStreamId(String clientStreamId, PlatformOutboundInstruction instruction); void sendToClient(String context, String clientId, PlatformOutboundInstruction instruction); @EventListener void on(PauseEventProcessorRequest evt); @EventListener void on(StartEventProcessorRequest evt); @EventListener void on(ApplicationDisconnected event); void onInboundInstruction(RequestCase requestCase, InstructionConsumer consumer); @EventListener void on(ProcessorStatusRequest event); @EventListener void on(ApplicationInactivityTimeout evt); Set<ClientComponent> getConnectedClients(); void requestReconnectForContext(String context); }
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()) { 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()); } }; } PlatformService(Topology topology,
                           ContextProvider contextProvider,
                           ClientIdRegistry clientIdRegistry,
                           ApplicationEventPublisher eventPublisher,
                           @Qualifier("platformInstructionAckSource")
                                   InstructionAckSource<PlatformOutboundInstruction> instructionAckSource); @Override void getPlatformServer(ClientIdentification request, StreamObserver<PlatformInfo> responseObserver); @Override StreamObserver<PlatformInboundInstruction> openStream(
            StreamObserver<PlatformOutboundInstruction> responseObserver); boolean requestReconnect(ClientComponent clientName); boolean requestReconnect(String clientId); void sendToAllClients(PlatformOutboundInstruction instruction); void sendToClientStreamId(String clientStreamId, PlatformOutboundInstruction instruction); void sendToClient(String context, String clientId, PlatformOutboundInstruction instruction); @EventListener void on(PauseEventProcessorRequest evt); @EventListener void on(StartEventProcessorRequest evt); @EventListener void on(ApplicationDisconnected event); void onInboundInstruction(RequestCase requestCase, InstructionConsumer consumer); @EventListener void on(ProcessorStatusRequest event); @EventListener void on(ApplicationInactivityTimeout evt); Set<ClientComponent> getConnectedClients(); void requestReconnectForContext(String context);  }