@Test
public void placeholderReplacement() throws Exception {
    final Target target = TestConstants.Targets.TARGET_WITH_PLACEHOLDER
            .withAddress("PATCH:" + TestConstants.Targets.TARGET_WITH_PLACEHOLDER.getAddress());
    connection = connection.toBuilder().setTargets(singletonList(target)).build();

    new TestKit(actorSystem) {{
        // GIVEN: local HTTP connection is connected
        final ActorRef underTest = actorSystem.actorOf(createClientActor(getRef(), getConnection(false)));
        underTest.tell(OpenConnection.of(connection.getId(), DittoHeaders.empty()), getRef());
        expectMsg(new Status.Success(BaseClientState.CONNECTED));

        // WHEN: a thing event is sent to a target with header mapping content-type=application/json
        final ThingModifiedEvent thingModifiedEvent = TestConstants.thingModified(Collections.emptyList());
        final OutboundSignal outboundSignal =
                OutboundSignalFactory.newOutboundSignal(thingModifiedEvent, singletonList(target));
        underTest.tell(outboundSignal, getRef());

        // THEN: a POST-request is forwarded to the path defined in the target
        final HttpRequest thingModifiedRequest = requestQueue.take();
        responseQueue.offer(HttpResponse.create().withStatus(StatusCodes.OK));
        assertThat(thingModifiedRequest.getUri().getPathString()).isEqualTo("/target:ditto/[email protected]");
    }};
}
 

/**
 * Creates the handler for messages in testing state. Overwrite and extend by additional matchers.
 *
 * @return an FSM function builder
 */
protected FSMStateFunctionBuilder<BaseClientState, BaseClientData> inTestingState() {
    return matchEvent(Status.Status.class, (e, d) -> Objects.equals(getSender(), getSelf()),
            (status, data) -> {
                log.info("{} status: <{}>", stateName(), status);
                data.getSessionSenders().forEach(sender ->
                        sender.first().tell(getStatusToReport(status, sender.second()), getSelf()));
                return stop();
            })
            .eventEquals(StateTimeout(), BaseClientData.class, (stats, data) -> {
                log.info("test timed out.");
                data.getSessionSenders().forEach(sender -> {
                    final DittoRuntimeException error = ConnectionFailedException.newBuilder(connectionId())
                            .description(String.format("Failed to open requested connection within <%d> seconds!",
                                    clientConfig.getTestingTimeout().getSeconds()))
                            .dittoHeaders(sender.second())
                            .build();
                    sender.first().tell(new Status.Failure(error), getSelf());
                });
                return stop();
            });
}
 

/**
 * Add meaningful message to status for reporting.
 *
 * @param status status to report.
 * @return status with meaningful message.
 */
private Status.Status getStatusToReport(final Status.Status status, final DittoHeaders dittoHeaders) {
    final Status.Status answerToPublish;
    if (status instanceof Status.Failure) {
        final Status.Failure failure = (Status.Failure) status;
        if (!(failure.cause() instanceof DittoRuntimeException)) {
            final DittoRuntimeException error = ConnectionFailedException.newBuilder(connectionId())
                    .description(describeEventualCause(failure.cause()))
                    .dittoHeaders(dittoHeaders)
                    .build();
            answerToPublish = new Status.Failure(error);
        } else {
            answerToPublish = status;
        }
    } else {
        answerToPublish = status;
    }
    return answerToPublish;
}
 

private CompletionStage<Status.Status> connectViaProxy(final String hostWithoutLookup, final int port) {
    final HttpProxyConfig httpProxyConfig = this.httpPushConfig.getHttpProxyConfig();
    try (final Socket proxySocket = new Socket(httpProxyConfig.getHostname(), httpProxyConfig.getPort())) {
        String proxyConnect = "CONNECT " + hostWithoutLookup + ":" + port + " HTTP/1.1\n";
        proxyConnect += "Host: " + hostWithoutLookup + ":" + port;

        if (!httpProxyConfig.getUsername().isEmpty()) {
            final String proxyUserPass = httpProxyConfig.getUsername() + ":" + httpProxyConfig.getPassword();
            proxyConnect += "\nProxy-Authorization: Basic " +
                    Base64.getEncoder().encodeToString(proxyUserPass.getBytes());
        }
        proxyConnect += "\n\n";
        proxySocket.getOutputStream().write(proxyConnect.getBytes());

        return checkProxyConnection(hostWithoutLookup, port, proxySocket);
    } catch (final Exception error) {
        return statusFailureFuture(new SocketException("Failed to connect to HTTP proxy: " + error.getMessage()));
    }
}
 

private void createMessageConsumer(final AmqpConsumerActor.CreateMessageConsumer command) {
    final Throwable error;
    if (currentSession != null) {
        // create required consumer
        final ConsumerData consumerData = command.getConsumerData();
        final ConsumerData newConsumerData =
                createJmsConsumer(currentSession, new HashMap<>(), consumerData.getSource(),
                        consumerData.getAddress(), consumerData.getAddressWithIndex());
        if (newConsumerData != null) {
            final Object response = command.toResponse(newConsumerData.getMessageConsumer());
            getSender().tell(response, getSelf());
            error = null;
        } else {
            error = new IllegalStateException("Failed to create message consumer");
        }
    } else {
        error = new IllegalStateException("No session");
    }
    if (error != null) {
        getSender().tell(new Status.Failure(error), getSelf());
    }
}
 

@Override
protected CompletionStage<Status.Status> startPublisherActor() {
    final CompletableFuture<Status.Status> future = new CompletableFuture<>();
    stopChildActor(amqpPublisherActor);
    final String namePrefix = AmqpPublisherActor.ACTOR_NAME_PREFIX;
    if (jmsSession != null) {
        final Props props = AmqpPublisherActor.props(connection(), jmsSession,
                connectivityConfig.getConnectionConfig());
        amqpPublisherActor = startChildActorConflictFree(namePrefix, props);
        future.complete(DONE);
    } else {
        future.completeExceptionally(ConnectionFailedException
                .newBuilder(connectionId())
                .message("Could not start publisher actor due to missing JMS session or connection!")
                .build());
    }
    return future;
}
 

@Override
protected void preEnhancement(final ReceiveBuilder receiveBuilder) {
    receiveBuilder
            // Incoming messages are handled in a separate stream parallelized by this actor's own dispatcher
            .match(ExternalMessage.class, this::handleInboundMessage)
            .match(Acknowledgement.class, acknowledgement ->
                    potentiallyForwardToAckregator(acknowledgement, () ->
                            handleNotExpectedAcknowledgement(acknowledgement))
            )
            .match(ThingCommandResponse.class, response -> {
                final ActorContext context = getContext();
                potentiallyForwardToAckregator(response,
                        () -> handleCommandResponse(response, null, context.getSender()));
            })
            // Outgoing responses and signals go through the signal enrichment stream
            .match(CommandResponse.class, response -> handleCommandResponse(response, null, getSender()))
            .match(Signal.class, signal -> handleSignal(signal, getSender()))
            .match(Status.Failure.class, f -> logger.warning("Got failure with cause {}: {}",
                    f.cause().getClass().getSimpleName(), f.cause().getMessage()));
}
 

@Test
public void testSubscribeFails() {
    new TestKit(actorSystem) {{
        final MockHiveMqtt5ClientFactory clientFactory = mockHiveMqtt5ClientFactory
                .withTestProbe(getRef())
                .withFailingSubscribe();

        final Props props =
                HiveMqtt5ClientActor.props(connection, getRef(), clientFactory, mockConnectionActor.ref());
        final ActorRef mqttClientActor = actorSystem.actorOf(props, "mqttClientActor-testSubscribeFails");

        mqttClientActor.tell(OpenConnection.of(connectionId, DittoHeaders.empty()), getRef());
        expectMsgClass(Status.Failure.class);

        mqttClientActor.tell(CloseConnection.of(connectionId, DittoHeaders.empty()), getRef());
        expectMsg(DISCONNECTED_SUCCESS);
    }};
}
 

@Override
public Receive createReceive() {
    return receiveBuilder()
            .match(CreateConnection.class, cc -> {
                if (allowCreate) {
                    log.info("connection created");
                    this.allowCreate = false;
                    sender().tell(new Status.Success("mock"), getSelf());
                } else {
                    sender().tell(new Status.Failure(new IllegalStateException("error message")),
                            getSelf());
                }
            })
            .match(OpenConnection.class,
                    oc -> sender().tell(new Status.Failure(new IllegalStateException("error message")),
                            getSelf()))
            .match(CloseConnection.class,
                    cc -> sender().tell(new Status.Failure(new IllegalStateException("error message")),
                            getSelf()))
            .match(DeleteConnection.class,
                    dc -> sender().tell(new Status.Failure(new IllegalStateException("error message")),
                            getSelf()))
            .build();
}
 

@Override
public void onReceive(Object message) throws Exception {
	if (message instanceof QueryState) {
		@SuppressWarnings("unchecked")
		QueryState<K> queryState = (QueryState<K>) message;

		LOG.debug("Received QueryState for key " + queryState.getKey() + ".");

		try {
			V value = keyValueState.getValue(queryState.getTimestamp(), queryState.getKey());

			if (value == null) {
				sender().tell(new StateNotFound<>(queryState.getKey()), getSelf());
			} else {
				sender().tell(new StateFound<>(queryState.getKey(), value), getSelf());
			}
		} catch (WrongKeyPartitionException ex) {
			sender().tell(new Status.Failure(ex), getSelf());
		}

		LOG.debug("Handled QueryState for key " + queryState.getKey() + ".");
	}else {
		throw new RuntimeException("Unknown message " + message);
	}
}
 

@Override
protected FSMStateFunctionBuilder<BaseClientState, BaseClientData> inTestingState() {
    return super.inTestingState()
            .event(Status.Status.class, (e, d) -> !Objects.equals(getSender(), getSelf()),
                    (status, data) -> handleStatusReportFromChildren(status))
            .event(ClientConnected.class, BaseClientData.class, (event, data) -> {
                final String url = data.getConnection().getUri();
                final String message = "Kafka connection to " + url + " established successfully";
                completeTestConnectionFuture(new Status.Success(message));
                return stay();
            })
            .event(ConnectionFailure.class, BaseClientData.class, (event, data) -> {
                completeTestConnectionFuture(new Status.Failure(event.getFailure().cause()));
                return stay();
            });
}
 

@Test
public void reconnectsInConnectingStateIfFailureResponseReceived() {
    new TestKit(actorSystem) {{
        final ConnectionId randomConnectionId = TestConstants.createRandomConnectionId();
        final Connection connection =
                TestConstants.createConnection(randomConnectionId, new Target[0]);
        final Props props = DummyClientActor.props(connection, getRef(), getRef(), getRef(), delegate);

        final ActorRef dummyClientActor = watch(actorSystem.actorOf(props));

        whenOpeningConnection(dummyClientActor, OpenConnection.of(randomConnectionId, DittoHeaders.empty()),
                getRef());
        andConnectionNotSuccessful(dummyClientActor);

        expectMsgClass(Status.Failure.class);

        thenExpectConnectClientCalled();
        thenExpectConnectClientCalledAfterTimeout(connectivityConfig.getClientConfig().getConnectingMinTimeout());
    }};
}
 

@Test
public void handlesCloseConnectionInConnectingState() {
    new TestKit(actorSystem) {{
        final ConnectionId randomConnectionId = TestConstants.createRandomConnectionId();
        final Connection connection =
                TestConstants.createConnection(randomConnectionId, new Target[0]);
        final Props props = DummyClientActor.props(connection, getRef(), getRef(), getRef(), delegate);

        final ActorRef dummyClientActor = watch(actorSystem.actorOf(props));

        whenOpeningConnection(dummyClientActor, OpenConnection.of(randomConnectionId, DittoHeaders.empty()),
                getRef());
        thenExpectConnectClientCalled();

        andClosingConnection(dummyClientActor, CloseConnection.of(randomConnectionId, DittoHeaders.empty()),
                getRef());
        thenExpectDisconnectClientCalled();
        andDisconnectionSuccessful(dummyClientActor, getRef());

        expectMsg(new Status.Success(BaseClientState.DISCONNECTED));
    }};
}
 

private void handleHandshakeMessage(RemoteHandshakeMessage handshakeMessage) {
	if (!isCompatibleVersion(handshakeMessage.getVersion())) {
		sendErrorIfSender(new AkkaHandshakeException(
			String.format(
				"Version mismatch between source (%s) and target (%s) rpc component. Please verify that all components have the same version.",
				handshakeMessage.getVersion(),
				getVersion())));
	} else if (!isGatewaySupported(handshakeMessage.getRpcGateway())) {
		sendErrorIfSender(new AkkaHandshakeException(
			String.format(
				"The rpc endpoint does not support the gateway %s.",
				handshakeMessage.getRpcGateway().getSimpleName())));
	} else {
		getSender().tell(new Status.Success(HandshakeSuccessMessage.INSTANCE), getSelf());
	}
}
 

@Test
public void shouldReconnectIfSocketIsClosed() {
    new TestKit(actorSystem) {{
        final ConnectionId randomConnectionId = TestConstants.createRandomConnectionId();
        final Connection connection =
                TestConstants.createConnection(randomConnectionId, new Target[0])
                        .toBuilder()
                        .uri("amqps://username:[email protected]:65536") // port 65536 does not even exist ;)
                        .build();
        final Props props = DummyClientActor.props(connection, getRef(), getRef(), getRef(), delegate);

        final ActorRef dummyClientActor = watch(actorSystem.actorOf(props));

        whenOpeningConnection(dummyClientActor, OpenConnection.of(randomConnectionId, DittoHeaders.empty()),
                getRef());

        expectMsgClass(Status.Failure.class);

        thenExpectCleanupResourcesCalled();
        Mockito.clearInvocations(delegate);
        thenExpectCleanupResourcesCalledAfterTimeout(
                connectivityConfig.getClientConfig().getConnectingMinTimeout());
        thenExpectNoConnectClientCalled();
    }};
}
 

@Test
public void doesNotReconnectIfConnectionSuccessful() {
    new TestKit(actorSystem) {{
        final ConnectionId randomConnectionId = TestConstants.createRandomConnectionId();
        final Connection connection =
                TestConstants.createConnection(randomConnectionId, new Target[0]);
        final Props props = DummyClientActor.props(connection, getRef(), getRef(), getRef(), delegate);

        final ActorRef dummyClientActor = watch(actorSystem.actorOf(props));

        whenOpeningConnection(dummyClientActor, OpenConnection.of(randomConnectionId, DittoHeaders.empty()),
                getRef());
        thenExpectConnectClientCalled();
        Mockito.clearInvocations(delegate);
        andConnectionSuccessful(dummyClientActor, getRef());

        expectMsgClass(Status.Success.class);
        thenExpectNoConnectClientCalledAfterTimeout(connectivityConfig.getClientConfig().getConnectingMinTimeout());
    }};
}
 

private void sendSyncResponse(Object response, String methodName) {
	if (isRemoteSender(getSender())) {
		Either<SerializedValue<?>, AkkaRpcException> serializedResult = serializeRemoteResultAndVerifySize(response, methodName);

		if (serializedResult.isLeft()) {
			getSender().tell(new Status.Success(serializedResult.left()), getSelf());
		} else {
			getSender().tell(new Status.Failure(serializedResult.right()), getSelf());
		}
	} else {
		getSender().tell(new Status.Success(response), getSelf());
	}
}
 

/**
 * Handle asynchronous {@link Callable}. This method simply executes the given {@link Callable}
 * in the context of the actor thread.
 *
 * @param callAsync Call async message
 */
private void handleCallAsync(CallAsync callAsync) {
	try {
		Object result = callAsync.getCallable().call();

		getSender().tell(new Status.Success(result), getSelf());
	} catch (Throwable e) {
		getSender().tell(new Status.Failure(e), getSelf());
	}
}
 

@Test
public void itShouldParseArticleTest() throws Exception {
    Future f = ask(askDemoActor, new ParseArticle(("http://www.google.com")), timeout);
    cacheProbe.expectMsgClass(GetRequest.class);
    cacheProbe.reply(new Status.Failure(new Exception("no cache")));

    httpClientProbe.expectMsgClass(String.class);
    httpClientProbe.reply(new HttpResponse(Articles.article1));

    String result = (String) Await.result(f, timeout.duration());
    assert(result.contains("I’ve been writing a lot in emacs lately"));
    assert(!result.contains("<body>"));
}
 

@Override
public void handleMessage(Object message) throws Exception {
	if (message instanceof ScheduleOrUpdateConsumers) {
		getSender().tell(
			decorateMessage(
				new Status.Failure(new Exception("Could not schedule or update consumers."))),
			getSelf());
	} else {
		super.handleMessage(message);
	}
}
 

private ArticleParseActor() {
    receive(ReceiveBuilder.
            match(String.class, html -> {
                        System.out.println(ArticleParser.apply(html).orElse("empty?"));

                        ArticleParser.apply(html).
                                onSuccess(body -> sender().tell(body, self())).
                                onFailure(t -> sender().tell(new Status.Failure(t), self()));
                    }
            ).
            matchAny(x -> {
                System.out.println("GOT A MSG!!! " + x);
            }).
            build());
}
 

private void sendSyncResponse(Object response, String methodName) {
	if (isRemoteSender(getSender())) {
		Either<SerializedValue<?>, AkkaRpcException> serializedResult = serializeRemoteResultAndVerifySize(response, methodName);

		if (serializedResult.isLeft()) {
			getSender().tell(new Status.Success(serializedResult.left()), getSelf());
		} else {
			getSender().tell(new Status.Failure(serializedResult.right()), getSelf());
		}
	} else {
		getSender().tell(new Status.Success(response), getSelf());
	}
}
 

private void sendSyncResponse(Object response, String methodName) {
	if (isRemoteSender(getSender())) {
		Either<SerializedValue<?>, AkkaRpcException> serializedResult = serializeRemoteResultAndVerifySize(response, methodName);

		if (serializedResult.isLeft()) {
			getSender().tell(new Status.Success(serializedResult.left()), getSelf());
		} else {
			getSender().tell(new Status.Failure(serializedResult.right()), getSelf());
		}
	} else {
		getSender().tell(new Status.Success(response), getSelf());
	}
}
 

@Override
public Receive createReceive() {
    return ReceiveBuilder.create()
            .match(StartChildActor.class, this::startChildActor)
            .match(Status.Failure.class, f -> log.error(f.cause(), "Got failure <{}>!", f))
            .matchAny(m -> {
                log.warning("Unknown message <{}>.", m);
                unhandled(m);
            })
            .build();
}
 

@Override
public Receive createReceive() {
    return ReceiveBuilder.create()
            .match(Subscribe.class, this::subscribe)
            .match(Unsubscribe.class, this::unsubscribe)
            .match(Terminated.class, this::terminated)
            .match(RemoveSubscriber.class, this::removeSubscriber)
            .matchEquals(Clock.TICK, this::tick)
            .match(SubscriptionsReader.class, this::updateSuccess)
            .match(Status.Failure.class, this::updateFailure)
            .matchAny(this::logUnhandled)
            .build();
}
 

private void updateFailure(final Status.Failure failure) {
    log.error(failure.cause(), "updateFailure");

    // try again next clock tick
    localSubscriptionsChanged = true;
    state = State.WAITING;
}
 

@Override
public AbstractActor.Receive createReceive() {
    return ReceiveBuilder.create()
            .match(Status.Failure.class, failure -> {
                Throwable cause = failure.cause();
                if (cause instanceof JsonRuntimeException) {
                    // wrap JsonRuntimeExceptions
                    cause = new DittoJsonException((RuntimeException) cause);
                }
                if (cause instanceof DittoRuntimeException) {
                    handleDittoRuntimeException((DittoRuntimeException) cause);
                } else if (cause instanceof EntityStreamSizeException) {
                    logger.warning("Got EntityStreamSizeException when a 'Command' was expected which means that" +
                            " the max. allowed http payload size configured in Akka was overstepped in this" +
                            " request.");
                    completeWithResult(
                            HttpResponse.create().withStatus(HttpStatusCode.REQUEST_ENTITY_TOO_LARGE.toInt()));
                } else {
                    logger.error(cause, "Got unknown Status.Failure when a 'Command' was expected.");
                    completeWithResult(
                            HttpResponse.create().withStatus(HttpStatusCode.INTERNAL_SERVER_ERROR.toInt()));
                }
            })
            .match(Whoami.class, this::handleWhoami)
            .match(DittoRuntimeException.class, this::handleDittoRuntimeException)
            .match(ReceiveTimeout.class,
                    receiveTimeout -> handleDittoRuntimeException(GatewayServiceUnavailableException.newBuilder()
                            .dittoHeaders(DittoHeaders.empty())
                            .build()))
            .match(Command.class, command -> !isResponseRequired(command), this::handleCommandWithoutResponse)
            .match(ThingModifyCommand.class, this::handleThingModifyCommand)
            .match(MessageCommand.class, this::handleMessageCommand)
            .match(Command.class, command -> handleCommandWithResponse(command,
                    getResponseAwaitingBehavior(getTimeoutExceptionSupplier(command))))
            .matchAny(m -> {
                logger.warning("Got unknown message, expected a 'Command': {}", m);
                completeWithResult(HttpResponse.create().withStatus(HttpStatusCode.INTERNAL_SERVER_ERROR.toInt()));
            })
            .build();
}
 

@Test
public void testTestConnection() {
    new TestKit(actorSystem) {{
        final Props props =
                AmqpClientActor.propsForTests(connection, getRef(), getRef(),
                        (ac, el) -> mockConnection);
        final ActorRef amqpClientActor = actorSystem.actorOf(props);

        amqpClientActor.tell(TestConnection.of(connection, DittoHeaders.empty()), getRef());
        expectMsgClass(Status.Success.class);
    }};
}
 

@Test
public void testTestConnectionFails() {
    new TestKit(actorSystem) {{
        final Props props = createFailingClientActor(getRef());
        final ActorRef mqttClientActor = actorSystem.actorOf(props);

        mqttClientActor.tell(TestConnection.of(connection, DittoHeaders.empty()), getRef());

        final Status.Failure failure = expectMsgClass(Status.Failure.class);
        assertThat(failure.cause()).isInstanceOf(ConnectionFailedException.class);

        expectDisconnectCalled();
    }};
}
 

@Test
public void testCreateConsumerFails() throws JMSException {
    new TestKit(actorSystem) {{
        doReturn(mockSession).when(mockConnection).createSession(Session.CLIENT_ACKNOWLEDGE);
        when(mockSession.createConsumer(any())).thenThrow(JMS_EXCEPTION);
        final Props props =
                AmqpClientActor.propsForTests(connection, getRef(),
                        getRef(), (ac, el) -> mockConnection);
        final ActorRef amqpClientActor = actorSystem.actorOf(props);

        amqpClientActor.tell(OpenConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef());
        expectMsgClass(Status.Failure.class);
    }};
}