This is untested. Treat it as a proof of concept.
Change-Id: Ieeef7c9481324984c9772b216d001254dec11ae9
Issue-ID: DCAEGEN2-1368
Signed-off-by: Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
<artifactId>mockito-core</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ </dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
--- /dev/null
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019 Nokia. All rights reserved.
+ * =========================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=====================================
+ */
+
+package org.onap.dcaegen2.services.sdk.rest.services.adapters.http;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since April 2019
+ */
+public final class HttpHeaders {
+
+ private HttpHeaders() {
+ }
+
+ public static final String CONTENT_TYPE = "Content-Type";
+ public static final String CONTENT_LENGTH = "Content-Length";
+}
package org.onap.dcaegen2.services.sdk.rest.services.adapters.http;
-import io.netty.buffer.ByteBuf;
import io.vavr.collection.HashMap;
import io.vavr.collection.Map;
-import java.util.function.BiFunction;
import org.immutables.value.Value;
import org.jetbrains.annotations.Nullable;
import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
-import org.reactivestreams.Publisher;
-import reactor.core.publisher.Mono;
-import reactor.netty.NettyOutbound;
-import reactor.netty.http.client.HttpClientRequest;
/**
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
HttpMethod method();
+ @Nullable RequestBody body();
+
@Value.Default
default RequestDiagnosticContext diagnosticContext() {
return RequestDiagnosticContext.create();
return HashMap.empty();
}
- @Value.Default
- default Publisher<ByteBuf> body() {
- return Mono.empty();
- }
-
@Value.Derived
default Map<String, String> headers() {
final RequestDiagnosticContext ctx = diagnosticContext();
import com.google.gson.JsonElement;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.Unpooled;
+import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
+import org.immutables.value.Value;
+import org.jetbrains.annotations.Nullable;
import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.netty.ByteBufFlux;
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
* @since March 2019
*/
-public final class RequestBody {
+@Value.Immutable
+public interface RequestBody {
- private RequestBody() {
+ Publisher<ByteBuf> contents();
+
+ @Nullable Integer length();
+
+ static RequestBody chunkedFromString(Publisher<String> contents) {
+ return chunkedFromString(contents, StandardCharsets.UTF_8);
}
- public static Publisher<ByteBuf> fromString(String contents) {
+ static RequestBody chunkedFromString(Publisher<String> contents, Charset charset) {
+ return ImmutableRequestBody.builder()
+ .length(null)
+ .contents(ByteBufFlux.fromString(contents, charset, ByteBufAllocator.DEFAULT))
+ .build();
+ }
+
+ static RequestBody fromString(String contents) {
return fromString(contents, StandardCharsets.UTF_8);
}
- public static Publisher<ByteBuf> fromString(String contents, Charset charset) {
- return ByteBufFlux.fromString(Mono.just(contents), charset, ByteBufAllocator.DEFAULT);
+ static RequestBody fromString(String contents, Charset charset) {
+ ByteBuf encodedContents = ByteBufAllocator.DEFAULT.buffer();
+ encodedContents.writeCharSequence(contents, charset);
+
+ return ImmutableRequestBody.builder()
+ .length(encodedContents.readableBytes())
+ .contents(Mono.just(encodedContents.retain()))
+ .build();
}
- public static Publisher<ByteBuf> fromJson(JsonElement contents) {
+ static RequestBody fromJson(JsonElement contents) {
return fromString(contents.toString());
}
}
ResponseReceiver<?> prepareRequest(HttpRequest request) {
- return httpClient
+ final HttpClient theClient = httpClient
.doOnRequest((req, conn) -> logRequest(request.diagnosticContext(), req))
.doOnResponse((rsp, conn) -> logResponse(request.diagnosticContext(), rsp))
- .headers(hdrs -> request.headers().forEach(hdr -> hdrs.set(hdr._1, hdr._2)))
+ .headers(hdrs -> request.headers().forEach(hdr -> hdrs.set(hdr._1, hdr._2)));
+
+ return prepareBody(request, theClient);
+ }
+
+ private ResponseReceiver<?> prepareBody(HttpRequest request, HttpClient theClient) {
+ if (request.body() == null) {
+ return prepareBodyWithoutContents(request, theClient);
+ } else {
+ return request.body().length() == null
+ ? prepareBodyChunked(request, theClient)
+ : prepareBodyUnchunked(request, theClient);
+ }
+ }
+
+ private ResponseReceiver<?> prepareBodyChunked(HttpRequest request, HttpClient theClient) {
+ return theClient
+ .chunkedTransfer(true)
+ .request(request.method().asNetty())
+ .send(request.body().contents())
+ .uri(request.url());
+ }
+
+ private ResponseReceiver<?> prepareBodyUnchunked(HttpRequest request, HttpClient theClient) {
+ return theClient
+ .headers(hdrs -> hdrs.set(HttpHeaders.CONTENT_LENGTH, request.body().length().toString()))
.request(request.method().asNetty())
- .send(request.body())
+ .send(request.body().contents())
.uri(request.url());
+ }
+ private ResponseReceiver<?> prepareBodyWithoutContents(HttpRequest request, HttpClient theClient) {
+ return theClient
+ .request(request.method().asNetty())
+ .uri(request.url());
}
private void logRequest(RequestDiagnosticContext context, HttpClientRequest httpClientRequest) {
package org.onap.dcaegen2.services.sdk.rest.services.adapters.http;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test.DummyHttpServer.sendString;
import io.netty.handler.codec.http.HttpResponseStatus;
httpServer = DummyHttpServer.start(routes ->
routes.get("/sample-get", (req, resp) -> sendString(resp, Mono.just("OK")))
.get("/sample-get-500", (req, resp) -> resp.status(HttpResponseStatus.INTERNAL_SERVER_ERROR).send())
+ .post("/headers-post", (req, resp) -> resp
+ .sendString(Mono.just(req.requestHeaders().toString())))
.post("/echo-post", (req, resp) -> resp.send(req.receive().retain()))
);
}
.expectComplete()
.verify(TIMEOUT);
}
+
+ @Test
+ void testChunkedEncoding() throws Exception {
+ // given
+ final String requestBody = "hello world";
+ final HttpRequest httpRequest = requestFor("/headers-post")
+ .method(HttpMethod.POST)
+ .body(RequestBody.chunkedFromString(Mono.just(requestBody)))
+ .build();
+
+ // when
+ final Mono<String> bodyAsString = cut.call(httpRequest)
+ .doOnNext(HttpResponse::throwIfUnsuccessful)
+ .map(HttpResponse::bodyAsString);
+
+ // then
+ StepVerifier.create(bodyAsString.map(String::toLowerCase))
+ .consumeNextWith(responseBody -> {
+ assertThat(responseBody).contains("transfer-encoding: chunked");
+ assertThat(responseBody).doesNotContain("content-length");
+ })
+ .expectComplete()
+ .verify(TIMEOUT);
+ }
+
+ @Test
+ void testUnchunkedEncoding() throws Exception {
+ // given
+ final String requestBody = "hello world";
+ final HttpRequest httpRequest = requestFor("/headers-post")
+ .method(HttpMethod.POST)
+ .body(RequestBody.fromString(requestBody))
+ .build();
+
+ // when
+ final Mono<String> bodyAsString = cut.call(httpRequest)
+ .doOnNext(HttpResponse::throwIfUnsuccessful)
+ .map(HttpResponse::bodyAsString);
+
+ // then
+ StepVerifier.create(bodyAsString.map(String::toLowerCase))
+ .consumeNextWith(responseBody -> {
+ assertThat(responseBody).doesNotContain("transfer-encoding");
+ assertThat(responseBody).contains("content-length");
+ })
+ .expectComplete()
+ .verify(TIMEOUT);
+ }
}
\ No newline at end of file
package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api;
+import com.google.gson.Gson;
+import io.vavr.Lazy;
+import org.jetbrains.annotations.NotNull;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClient;
import org.onap.dcaegen2.services.sdk.rest.services.annotations.ExperimentalApi;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.impl.MessageRouterClientImpl;
/**
+ * <b>WARNING</b>: This is a proof-of-concept. It is untested. API may change or be removed. Use at your own risk.
+ * You've been warned.
+ *
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
* @since 1.1.4
*/
@ExperimentalApi
public final class DmaapClientFactory {
+ private static final Lazy<MessageRouterClientImpl> THE_CLIENT = Lazy.of(() ->
+ new MessageRouterClientImpl(RxHttpClient.create(), new Gson()));
+
private DmaapClientFactory() {
}
- public static MessageRouterPublisher createMessageRouterPublisher() {
- throw new UnsupportedOperationException("not implemented yet");
+ public static @NotNull MessageRouterPublisher createMessageRouterPublisher() {
+ return THE_CLIENT.get();
}
- public static MessageRouterSubscriber createMessageRouterSubscriber() {
- throw new UnsupportedOperationException("not implemented yet");
+ public static @NotNull MessageRouterSubscriber createMessageRouterSubscriber() {
+ return THE_CLIENT.get();
}
}
--- /dev/null
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019 Nokia. All rights reserved.
+ * =========================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=====================================
+ */
+
+package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.impl;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import io.netty.buffer.ByteBuf;
+import io.vavr.collection.HashMap;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import org.jetbrains.annotations.NotNull;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpHeaders;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpMethod;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpResponse;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.ImmutableHttpRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RequestBody;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClient;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterPublisher;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterSubscriber;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishResponse;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterSubscribeResponse;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishResponse;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeResponse;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since March 2019
+ */
+// TODO: This is a PoC. It's untested.
+public class MessageRouterClientImpl implements MessageRouterPublisher, MessageRouterSubscriber {
+
+ private static final Duration WINDOW_MAX_TIME = Duration.ofSeconds(1);
+ private static final int WINDOW_MAX_SIZE = 512;
+ private final RxHttpClient httpClient;
+ private final Gson gson;
+
+ public MessageRouterClientImpl(RxHttpClient httpClient, Gson gson) {
+ this.httpClient = httpClient;
+ this.gson = gson;
+ }
+
+ @Override
+ public Flux<MessageRouterPublishResponse> put(
+ MessageRouterPublishRequest request,
+ Flux<? extends JsonElement> items) {
+ return items.windowTimeout(WINDOW_MAX_SIZE, WINDOW_MAX_TIME).flatMap(subItems ->
+ subItems.collect(JsonArray::new, JsonArray::add)
+ .filter(arr -> arr.size() > 0)
+ .map(RequestBody::fromJson)
+ .flatMap(body -> httpClient.call(buildPostHttpRequest(request, body)))
+ .map(this::buildPutResponse));
+ }
+
+ @Override
+ public Mono<MessageRouterSubscribeResponse> get(MessageRouterSubscribeRequest request) {
+ return httpClient.call(buildGetHttpRequest(request)).map(this::buildGetResponse);
+ }
+
+ private @NotNull MessageRouterPublishResponse buildPutResponse(HttpResponse httpResponse) {
+ final ImmutableMessageRouterPublishResponse.Builder builder =
+ ImmutableMessageRouterPublishResponse.builder();
+ return httpResponse.successful()
+ ? builder.build()
+ : builder.failReason(extractFailReason(httpResponse)).build();
+ }
+
+ private @NotNull MessageRouterSubscribeResponse buildGetResponse(HttpResponse httpResponse) {
+ final ImmutableMessageRouterSubscribeResponse.Builder builder =
+ ImmutableMessageRouterSubscribeResponse.builder();
+ return httpResponse.successful()
+ ? builder.items(httpResponse.bodyAsJson(StandardCharsets.UTF_8, gson, JsonArray.class)).build()
+ : builder.failReason(extractFailReason(httpResponse)).build();
+ }
+
+ private String extractFailReason(HttpResponse httpResponse) {
+ return String.format("%d %s%n%s", httpResponse.statusCode(), httpResponse.statusReason(),
+ httpResponse.bodyAsString());
+ }
+
+ private @NotNull HttpRequest buildPostHttpRequest(MessageRouterPublishRequest request, RequestBody body) {
+ return ImmutableHttpRequest.builder()
+ .method(HttpMethod.POST)
+ .url(request.sinkDefinition().topicUrl())
+ .diagnosticContext(request.diagnosticContext())
+ .customHeaders(HashMap.of(HttpHeaders.CONTENT_TYPE, request.contentType()))
+ .body(body)
+ .build();
+ }
+
+ private @NotNull HttpRequest buildGetHttpRequest(MessageRouterSubscribeRequest request) {
+ return ImmutableHttpRequest.builder()
+ .method(HttpMethod.GET)
+ .url(buildSubscribeUrl(request))
+ .diagnosticContext(request.diagnosticContext())
+ .build();
+ }
+
+ private String buildSubscribeUrl(MessageRouterSubscribeRequest request) {
+ return String.format("%s/%s/%s", request.sourceDefinition().topicUrl(), request.consumerGroup(), request.consumerId());
+ }
+}
--- /dev/null
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019 Nokia. All rights reserved.
+ * =========================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=====================================
+ */
+
+package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model;
+
+import java.util.UUID;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since March 2019
+ */
+final class Constants {
+
+ private Constants() {
+ }
+
+ static final String CLASS_LOADER_SCOPED_UNIQUE_ID = UUID.randomUUID().toString();
+}
default RequestDiagnosticContext diagnosticContext() {
return RequestDiagnosticContext.create();
}
-
}
import org.immutables.value.Value;
import org.onap.dcaegen2.services.sdk.model.streams.dmaap.MessageRouterSink;
import org.onap.dcaegen2.services.sdk.rest.services.annotations.ExperimentalApi;
-import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
/**
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
public interface MessageRouterPublishRequest extends DmaapRequest {
MessageRouterSink sinkDefinition();
+
+ @Value.Default
+ default String contentType() {
+ return "application/json";
+ }
}
* limitations under the License.
* ============LICENSE_END=====================================
*/
-
package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model;
+import java.time.Duration;
import org.immutables.value.Value;
+import org.jetbrains.annotations.Nullable;
import org.onap.dcaegen2.services.sdk.model.streams.dmaap.MessageRouterSource;
import org.onap.dcaegen2.services.sdk.rest.services.annotations.ExperimentalApi;
-import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
-
/**
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
public interface MessageRouterSubscribeRequest extends DmaapRequest {
MessageRouterSource sourceDefinition();
+
+ String consumerGroup();
+
+ @Nullable Duration timeout();
+
+ default String consumerId() {
+ return Constants.CLASS_LOADER_SCOPED_UNIQUE_ID;
+ }
}
Code Review / dcaegen2 / services / sdk.git / commitdiff