@Nullable
private static MessageMapper createAnyMessageMapper(final Class<?> clazz,
        final DynamicAccess dynamicAccess) {
    final ClassTag<MessageMapper> tag = scala.reflect.ClassTag$.MODULE$.apply(MessageMapper.class);
    final Try<MessageMapper> mapperTry = dynamicAccess.createInstanceFor(clazz, List$.MODULE$.empty(), tag);

    if (mapperTry.isFailure()) {
        final Throwable error = mapperTry.failed().get();
        if (error instanceof ClassNotFoundException || error instanceof InstantiationException ||
                error instanceof ClassCastException) {
            return null;
        } else {
            throw new IllegalStateException("There was an unknown error when trying to creating instance for '"
                    + clazz + "'", error);
        }
    }

    return mapperTry.get();
}
 

@Override
public <T> Flow<Pair<HttpRequest, T>, Pair<Try<HttpResponse>, T>, ?> createFlow(final ActorSystem system,
        final LoggingAdapter log) {

    final Http http = Http.get(system);
    final ConnectionPoolSettings poolSettings = getConnectionPoolSettings(system);
    final Flow<Pair<HttpRequest, T>, Pair<Try<HttpResponse>, T>, ?> flow;
    if (null != httpsConnectionContext) {
        final ConnectHttp connectHttpsWithCustomSSLContext =
                ConnectHttp.toHostHttps(baseUri).withCustomHttpsContext(httpsConnectionContext);
        // explicitly added <T> as in (some?) IntelliJ idea the line would show an error:
        flow = http.<T>cachedHostConnectionPoolHttps(connectHttpsWithCustomSSLContext, poolSettings, log);
    } else {
        // explicitly added <T> as in (some?) IntelliJ idea the line would show an error:
        // no SSL, hence no need for SSLContextCreator
        flow = http.<T>cachedHostConnectionPool(ConnectHttp.toHost(baseUri), poolSettings, log);
    }
    return flow.buffer(parallelism, OverflowStrategy.backpressure());
}
 

@Test
public void basicAuth() throws Exception {
    // GIVEN: the connection has plain credentials in the URI
    connection = connection.toBuilder()
            .uri("http://username:[email protected]:" + binding.localAddress().getPort() + "/path/prefix/")
            .build();
    final HttpPushFactory underTest = HttpPushFactory.of(connection, connectionConfig.getHttpPushConfig());
    final Pair<SourceQueueWithComplete<HttpRequest>, SinkQueueWithCancel<Try<HttpResponse>>> pair =
            newSourceSinkQueues(underTest);
    final SourceQueueWithComplete<HttpRequest> sourceQueue = pair.first();
    final SinkQueueWithCancel<Try<HttpResponse>> sinkQueue = pair.second();
    final HttpRequest request = underTest.newRequest(HttpPublishTarget.of("PUT:/path/appendage/"));
    final HttpResponse response = HttpResponse.create().withStatus(StatusCodes.OK);

    // WHEN: request-response cycle is carried out
    responseQueue.offer(CompletableFuture.completedFuture(response));
    sourceQueue.offer(request);
    final HttpRequest actualRequest = requestQueue.take();

    // THEN: actual received request has a basic auth header
    assertThat(actualRequest.getHeader(Authorization.class))
            .contains(Authorization.basic("username", "password"));
}
 

private Pair<SourceQueueWithComplete<HttpRequest>, SinkQueueWithCancel<Try<HttpResponse>>> newSourceSinkQueues(
        final HttpPushFactory underTest) {

    return Source.<HttpRequest>queue(10, OverflowStrategy.dropNew())
            .map(r -> Pair.create(r, null))
            .viaMat(underTest.createFlow(actorSystem, actorSystem.log()), Keep.left())
            .map(Pair::first)
            .toMat(Sink.queue(), Keep.both())
            .run(mat);
}
 

private List<Object> getInputFields(IndexedRecord inputRecord, String columnName) {
    // Adapt non-avpath syntax to avpath.
    // TODO: This should probably not be automatic, use the actual syntax.
    if (!columnName.startsWith("."))
        columnName = "." + columnName;
    Try<scala.collection.immutable.List<Evaluator.Ctx>> result = wandou.avpath.package$.MODULE$.select(inputRecord,
            columnName);
    List<Object> values = new ArrayList<Object>();
    if (result.isSuccess()) {
        for (Evaluator.Ctx ctx : JavaConversions.asJavaCollection(result.get())) {
            values.add(ctx.value());
        }
    } else {
        // Evaluating the expression failed, and we can handle the exception.
        Throwable t = result.failed().get();
        throw ProcessingErrorCode.createAvpathSyntaxError(t, columnName, -1);
    }
    return values;
}
 

/**
 * Use an AVPath to extract data from an indexed record
 * 
 * @param record an indexed record
 * @param avPath the path to elements to extract (can be one or multiples elements)
 * @return the extracted data as a list.
 */
public static List<Evaluator.Ctx> getInputFields(IndexedRecord record, String avPath) {
    // Adapt non-avpath syntax to avpath.
    // TODO: This should probably not be automatic, use the actual syntax.
    if (!avPath.startsWith("."))
        avPath = "." + avPath;
    Try<scala.collection.immutable.List<Evaluator.Ctx>> result =
            wandou.avpath.package$.MODULE$.select(record, avPath);
    List<Evaluator.Ctx> results = new ArrayList<Evaluator.Ctx>();
    if (result.isSuccess()) {
        for (Evaluator.Ctx ctx : JavaConversions.asJavaCollection(result.get())) {
            results.add(ctx);
        }
    } else {
        // Evaluating the expression failed, and we can handle the exception.
        throw ProcessingErrorCode.createAvpathSyntaxError(result.failed().get(), avPath, -1);
    }
    return results;
}
 

private Optional<MDNReport> parseReport(Entity report) {
    LOGGER.debug("Parsing report");
    try (InputStream inputStream = ((SingleBody) report.getBody()).getInputStream()) {
        Try<MDNReport> result = MDNReportParser.parse(inputStream, report.getCharset());
        if (result.isSuccess()) {
            return Optional.of(result.get());
        } else {
            LOGGER.error("unable to parse MESSAGE_DISPOSITION_NOTIFICATION part", result.failed().get());
            return Optional.empty();
        }
    } catch (IOException e) {
        LOGGER.error("unable to parse MESSAGE_DISPOSITION_NOTIFICATION part", e);
        return Optional.empty();
    }
}
 

@Override
public TypeSerializerSchemaCompatibility<Try<E>> resolveSchemaCompatibility(TypeSerializer<Try<E>> newSerializer) {

	return CompositeTypeSerializerUtil.delegateCompatibilityCheckToNewSnapshot(
		newSerializer,
		new ScalaTrySerializerSnapshot<>(),
		getNestedSerializersAndConfigs().get(0).f1,
		getNestedSerializersAndConfigs().get(1).f1);
}
 

public static <T> ComposableFuture<T> from(final FutureProvider<T> source) {
  return ComposableFutures.build(consumer -> {
    final Future<T> future = source.provide();
    future.onComplete(new AbstractFunction1<Try<T>, Void>() {
      @Override public Void apply(final Try<T> res) {
        if (res.isSuccess()) {
          consumer.consume(com.outbrain.ob1k.concurrent.Try.fromValue(res.get()));
        } else {
          consumer.consume(com.outbrain.ob1k.concurrent.Try.fromError(res.failed().get()));
        }

        return null;
      }
    }, ctx);
  });
}
 

@Override
public Try<MySQLConnection> validate(final MySQLConnection item) {
  if (!item.isConnected()) {
    return new Failure<>(new ConnectionNotConnectedException(item));
  }

  if (item.isQuerying()) {
    return new Failure<>(new ConnectionStillRunningQueryException(item.count(), false));
  }

  return new Success<>(item);
}
 

public static void main(String[] args){
 Try<BinaryXpr> parsed=XprParser.parseXpr("dico > 5");
 //Try<BinaryXpr> parsed=XprParser$.MODULE$.parseXpr("dfsfsdf");
 if (parsed.isFailure())
  System.out.println("abcdcdc "+parsed.failed().get());
 else System.out.println(parsed.get().toString());
}
 

@Override
public TypeSerializerSchemaCompatibility<Try<E>> resolveSchemaCompatibility(TypeSerializer<Try<E>> newSerializer) {

	return CompositeTypeSerializerUtil.delegateCompatibilityCheckToNewSnapshot(
		newSerializer,
		new ScalaTrySerializerSnapshot<>(),
		getNestedSerializersAndConfigs().get(0).f1,
		getNestedSerializersAndConfigs().get(1).f1);
}
 

public ScalaTrySerializerSnapshotMigrationTest(TestSpecification<Try<String>> testSpecification) {
	super(testSpecification);
}
 

@Override
public TypeSerializerSchemaCompatibility<Try<E>> resolveSchemaCompatibility(TypeSerializer<Try<E>> newSerializer) {

	return CompositeTypeSerializerUtil.delegateCompatibilityCheckToNewSnapshot(
		newSerializer,
		new ScalaTrySerializerSnapshot<>(),
		getNestedSerializersAndConfigs().get(0).f1,
		getNestedSerializersAndConfigs().get(1).f1);
}
 

public ScalaTrySerializerSnapshotMigrationTest(TestSpecification<Try<String>> testSpecification) {
	super(testSpecification);
}
 

public static <T> Future<T> scalaToVertx(scala.concurrent.Future<T> future, ExecutionContext ec) {
    Future<T> fut = Future.future();
    future.onComplete(new AbstractFunction1<Try<T>, Void>() {
        @Override
        public Void apply(Try<T> v1) {
            if (v1.isSuccess()) {
                fut.complete(v1.get());
            } else {
                fut.fail(v1.failed().get());
            }
            return null;
        }
    }, ec);
    return fut;
}
 

public static <T> Future<Void> scalaToVertxVoid(scala.concurrent.Future<T> future, ExecutionContext ec) {
    Future<Void> fut = Future.future();
    future.onComplete(new AbstractFunction1<Try<T>, Void>() {
        @Override
        public Void apply(Try<T> v1) {
            if (v1.isSuccess()) {
                fut.complete();
            } else {
                fut.fail(v1.failed().get());
            }
            return null;
        }
    }, ec);
    return fut;
}
 

public static <V> Function1<Try<V>, Void> toFunction1(Handler<AsyncResult<V>> code) {
    return new AbstractFunction1<Try<V>, Void>() {
        @Override
        public Void apply(Try<V> v1) {
            if (v1.isSuccess()) {
                code.handle(Future.succeededFuture(v1.get()));
            } else {
                code.handle(Future.failedFuture(v1.failed().get()));
            }
            return null;
        }
    };
}
 

@SuppressWarnings("unchecked")
public static void applyEnd(final Object thiz, final Object returned, final Throwable thrown) {
  final LocalSpanContext context = LocalSpanContext.get(COMPONENT_NAME);
  if (context == null)
    return;

  if (context.decrementAndGet() != 0)
    return;

  final Span span = context.getSpan();
  context.closeScope();

  if (thrown != null) {
    OpenTracingApiUtil.setErrorTag(span, thrown);
    span.finish();
    return;
  }

  ((Future<Result>)returned).onComplete(new Function1<Try<Result>,Object>() {
    @Override
    public Object apply(final Try<Result> response) {
      if (response.isFailure())
        OpenTracingApiUtil.setErrorTag(span, response.failed().get());
      else
        span.setTag(Tags.HTTP_STATUS, response.get().header().status());

      span.finish();
      return null;
    }
  }, ((Action<?>)thiz).executionContext());
}
 

/**
 * Loads the configured {@code ProtocolAdapterProvider} by reflection.
 * This calls the 1-argument constructor every subclass of ProtocolAdapterProvider should implement.
 *
 * @param protocolConfig provides the class name of the ProtocolAdapterProvider to be loaded.
 * @param actorSystem Akka actor system to perform reflection with.
 * @return the loaded protocol adapter provider.
 */
public static ProtocolAdapterProvider load(final ProtocolConfig protocolConfig, final ActorSystem actorSystem) {
    final String className = protocolConfig.getProviderClassName();
    final ClassTag<ProtocolAdapterProvider> tag = ClassTag$.MODULE$.apply(ProtocolAdapterProvider.class);
    final List<Tuple2<Class<?>, Object>> constructorArgs =
            Collections.singletonList(new Tuple2<>(ProtocolConfig.class, protocolConfig));
    final DynamicAccess dynamicAccess = ((ExtendedActorSystem) actorSystem).dynamicAccess();
    final Try<ProtocolAdapterProvider> providerBox = dynamicAccess.createInstanceFor(className,
            JavaConverters.asScalaBuffer(constructorArgs).toList(), tag);

    return providerBox.get();
}
 

/**
 * Depending on the request headers, one of the supported authentication mechanisms is applied.
 *
 * @param dittoHeaders the DittoHeaders containing already gathered context information.
 * @param inner the inner route which will be wrapped with the {@link DittoHeaders}.
 * @return the inner route.
 */
public Route authenticate(final DittoHeaders dittoHeaders, final Function<DittoHeadersBuilder<?, ?>, Route> inner) {
    return extractRequestContext(requestContext -> {
        final Uri requestUri = requestContext.getRequest().getUri();

        final CompletableFuture<AuthenticationResult> authenticationResult =
                authenticationChain.authenticate(requestContext, dittoHeaders);

        final Function<Try<AuthenticationResult>, Route> handleAuthenticationTry =
                authenticationResultTry -> handleAuthenticationTry(authenticationResultTry, requestUri,
                        dittoHeaders, inner);

        return Directives.onComplete(authenticationResult, handleAuthenticationTry);
    });
}
 

private Route handleAuthenticationTry(final Try<AuthenticationResult> authenticationResultTry,
        final Uri requestUri,
        final DittoHeaders dittoHeaders,
        final Function<DittoHeadersBuilder<?, ?>, Route> inner) {

    if (authenticationResultTry.isSuccess()) {
        final AuthenticationResult authenticationResult = authenticationResultTry.get();
        if (authenticationResult.isSuccess()) {
            return inner.apply(authenticationResult.getDittoHeaders().toBuilder());
        }
        return handleFailedAuthentication(authenticationResult.getReasonOfFailure(), requestUri, dittoHeaders);
    }
    return handleFailedAuthentication(authenticationResultTry.failed().get(), requestUri, dittoHeaders);
}
 

private void processResponse(final Pair<Try<HttpResponse>, HttpPushContext> responseWithMessage) {
    final Try<HttpResponse> tryResponse = responseWithMessage.first();
    final HttpPushContext context = responseWithMessage.second();
    final ExternalMessage message = context.getExternalMessage();
    final Uri requestUri = context.getRequestUri();
    if (tryResponse.isFailure()) {
        final Throwable error = tryResponse.toEither().left().get();
        final String errorDescription = MessageFormat.format("Failed to send HTTP request to <{0}>.",
                stripUserInfo(requestUri));
        log.debug("Failed to send message <{}> due to <{}>", message, error);
        responsePublishedMonitor.failure(message, errorDescription);
        escalate(error, errorDescription);
    } else {
        final HttpResponse response = tryResponse.toEither().right().get();
        log.debug("Sent message <{}>. Got response <{} {}>", message, response.status(), response.getHeaders());
        if (response.status().isSuccess()) {
            responsePublishedMonitor.success(message,
                    "HTTP call to <{0}> successfully responded with status <{1}>.",
                    stripUserInfo(requestUri), response.status());
            response.discardEntityBytes(materializer);
        } else {
            getResponseBody(response, materializer)
                    .thenAccept(body -> responsePublishedMonitor.failure(message,
                            "HTTP call to <{0}> responded with status <{1}> and body: {2}.",
                            stripUserInfo(requestUri),
                            response.status(), body)
                    )
                    .exceptionally(bodyReadError -> {
                        responsePublishedMonitor.failure(message,
                                "HTTP call to <{0}> responded with status <{1}>. Failed to read body within {2} ms",
                                stripUserInfo(requestUri), response.status(), READ_BODY_TIMEOUT_MS);
                        LogUtil.enhanceLogWithCorrelationId(log, message.getInternalHeaders());
                        log.info("Got <{}> when reading body of publish response to <{}>", bodyReadError,
                                message);
                        return null;
                    });
        }
    }
}
 

@Test
public void withHttpProxyConfig() throws Exception {
    // GIVEN: the connection's URI points to an unreachable host
    connection = connection.toBuilder()
            .uri("http://no.such.host.example:42/path/prefix/")
            .build();

    // GIVEN: the HTTP-push factory has the proxy configured to the test server binding
    final HttpPushFactory underTest = HttpPushFactory.of(connection, new HttpPushConfig() {
        @Override
        public int getMaxQueueSize() {
            return 0;
        }

        @Override
        public HttpProxyConfig getHttpProxyConfig() {
            return getEnabledProxyConfig(binding);
        }
    });
    final Pair<SourceQueueWithComplete<HttpRequest>, SinkQueueWithCancel<Try<HttpResponse>>> pair =
            newSourceSinkQueues(underTest);
    final SourceQueueWithComplete<HttpRequest> sourceQueue = pair.first();
    final SinkQueueWithCancel<Try<HttpResponse>> sinkQueue = pair.second();
    final HttpRequest request = underTest.newRequest(HttpPublishTarget.of("PUT:/path/appendage/"));

    // WHEN: request is made
    sourceQueue.offer(request);

    // THEN: CONNECT request is made to the Akka HTTP test server.
    // THEN: Akka HTTP server rejects CONNECT request, creating a failed response
    final Optional<Try<HttpResponse>> optionalTryResponse = sinkQueue.pull().toCompletableFuture().join();
    assertThat(optionalTryResponse).isNotEmpty();
    final Try<HttpResponse> tryResponse = optionalTryResponse.get();
    assertThat(tryResponse).isInstanceOf(Failure.class);
    assertThat(tryResponse.failed().get()).isInstanceOf(ProxyConnectionFailedException.class);
    assertThat(tryResponse.failed().get().getMessage())
            .contains("proxy rejected to open a connection to no.such.host.example:42 with status code: 400");
}
 

@Test
public void handleFailure() throws Exception {
    new TestKit(actorSystem) {{
        // GIVEN: An HTTP-push connection is established against localhost.
        final HttpPushFactory underTest = HttpPushFactory.of(connection, connectionConfig.getHttpPushConfig());
        final Pair<SourceQueueWithComplete<HttpRequest>, SinkQueueWithCancel<Try<HttpResponse>>> pair =
                newSourceSinkQueues(underTest);
        final SourceQueueWithComplete<HttpRequest> sourceQueue = pair.first();
        final SinkQueueWithCancel<Try<HttpResponse>> sinkQueue = pair.second();
        final HttpRequest request1 = underTest.newRequest(HttpPublishTarget.of("1"));
        final HttpRequest request2 = underTest.newRequest(HttpPublishTarget.of("2"));
        final HttpRequest request3 = underTest.newRequest(HttpPublishTarget.of("3"));
        final HttpResponse response1 = HttpResponse.create().withStatus(StatusCodes.IM_A_TEAPOT);
        final HttpResponse response3 = HttpResponse.create().withStatus(StatusCodes.BLOCKED_BY_PARENTAL_CONTROLS);

        // GIVEN: The connection is working.
        responseQueue.offer(CompletableFuture.completedFuture(response1));
        sourceQueue.offer(request1);
        assertThat(requestQueue.take().getUri()).isEqualTo(request1.getUri());
        final Try<HttpResponse> responseOrError1 = pullResponse(sinkQueue);
        assertThat(responseOrError1.isSuccess()).isTrue();
        assertThat(responseOrError1.get().status()).isEqualTo(response1.status());

        // WHEN: In-flight request is killed
        // THEN: Akka HTTP responds with status 500
        responseQueue.offer(new CompletableFuture<>());
        sourceQueue.offer(request2);
        assertThat(requestQueue.take().getUri()).isEqualTo(request2.getUri());
        shutdownAllServerStreams();
        final Try<HttpResponse> responseOrError2 = pullResponse(sinkQueue);
        assertThat(responseOrError2.isSuccess()).isTrue();
        assertThat(responseOrError2.get().status()).isEqualTo(StatusCodes.INTERNAL_SERVER_ERROR);

        // WHEN: HTTP server becomes available again.
        // THEN: A new request resumes and the previously failed request is discarded.
        responseQueue.offer(CompletableFuture.completedFuture(response3));
        sourceQueue.offer(request3);
        assertThat(requestQueue.take().getUri()).isEqualTo(request3.getUri());
        final Try<HttpResponse> responseOrError3 = pullResponse(sinkQueue);
        assertThat(responseOrError3.isSuccess()).isTrue();
        assertThat(responseOrError3.get().status()).isEqualTo(response3.status());
    }};
}
 

private static Try<HttpResponse> pullResponse(final SinkQueueWithCancel<Try<HttpResponse>> responseQueue) {
    return responseQueue.pull()
            .toCompletableFuture()
            .join()
            .orElseThrow(() -> new AssertionError("Response expected"));
}
 

@Override
public <T> Flow<Pair<HttpRequest, T>, Pair<Try<HttpResponse>, T>, ?> createFlow(final ActorSystem system,
        final LoggingAdapter log) {
    return Flow.<Pair<HttpRequest, T>>create()
            .map(pair -> Pair.create(Try.apply(() -> mapper.apply(pair.first())), pair.second()));
}
 

public ScalaTrySerializerUpgradeTest(TestSpecification<Try<String>, Try<String>> testSpecification) {
	super(testSpecification);
}
 

@Override
public TypeSerializer<Try<String>> createPriorSerializer() {
	return new TrySerializer<>(StringSerializer.INSTANCE, new ExecutionConfig());
}
 

@SuppressWarnings("unchecked")
@Override
public Try<String> createTestData() {
	return new Failure(new SpecifiedException("Specified exception for ScalaTry."));
}