From 97293ebf83bcbb1ba67e84cb04506bd1e707eaae Mon Sep 17 00:00:00 2001 From: Sergei Egorov Date: Fri, 14 Sep 2018 16:46:30 +0200 Subject: [PATCH 1/4] remove Netty classes from OkHttpInvocationBuilder --- .../okhttp/OkHttpInvocationBuilder.java | 143 ++++++++++++------ 1 file changed, 95 insertions(+), 48 deletions(-) diff --git a/core/src/main/java/org/testcontainers/dockerclient/transport/okhttp/OkHttpInvocationBuilder.java b/core/src/main/java/org/testcontainers/dockerclient/transport/okhttp/OkHttpInvocationBuilder.java index bc84e44a323..0fa919e3de4 100644 --- a/core/src/main/java/org/testcontainers/dockerclient/transport/okhttp/OkHttpInvocationBuilder.java +++ b/core/src/main/java/org/testcontainers/dockerclient/transport/okhttp/OkHttpInvocationBuilder.java @@ -2,7 +2,9 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.MappingIterator; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializationFeature; import com.github.dockerjava.api.async.ResultCallback; import com.github.dockerjava.api.exception.BadRequestException; import com.github.dockerjava.api.exception.ConflictException; @@ -13,13 +15,10 @@ import com.github.dockerjava.api.exception.NotModifiedException; import com.github.dockerjava.api.exception.UnauthorizedException; import com.github.dockerjava.api.model.Frame; +import com.github.dockerjava.api.model.StreamType; import com.github.dockerjava.core.InvocationBuilder; -import com.github.dockerjava.netty.handler.FramedResponseStreamHandler; -import com.github.dockerjava.netty.handler.JsonResponseCallbackHandler; -import io.netty.buffer.ByteBuf; -import io.netty.buffer.Unpooled; -import io.netty.channel.SimpleChannelInboundHandler; import lombok.AccessLevel; +import lombok.RequiredArgsConstructor; import lombok.SneakyThrows; import lombok.experimental.FieldDefaults; import lombok.extern.slf4j.Slf4j; @@ -34,13 +33,15 @@ import okio.BufferedSource; import okio.Okio; import okio.Source; +import org.testcontainers.DockerClientFactory; import javax.annotation.Nullable; import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; import java.lang.reflect.Field; -import java.util.concurrent.atomic.AtomicBoolean; +import java.util.Objects; +import java.util.function.Consumer; @Slf4j @FieldDefaults(level = AccessLevel.PRIVATE, makeFinal = true) @@ -91,7 +92,7 @@ public void get(ResultCallback resultCallback) { executeAndStream( request, resultCallback, - new FramedResponseStreamHandler(resultCallback) + new FramedSink(resultCallback) ); } @@ -187,7 +188,7 @@ public void run() { okHttpClient, request, resultCallback, - new FramedResponseStreamHandler(resultCallback) + new FramedSink(resultCallback) ); } @@ -200,7 +201,7 @@ public void post(TypeReference typeReference, ResultCallback resultCal executeAndStream( request, resultCallback, - new JsonResponseCallbackHandler<>(typeReference, resultCallback) + new JsonSink<>(typeReference, resultCallback) ); } @@ -247,7 +248,7 @@ public MediaType contentType() { @Override public void writeTo(BufferedSink sink) throws IOException { - try(Source source = Okio.source(body)) { + try (Source source = Okio.source(body)) { sink.writeAll(source); } } @@ -286,49 +287,95 @@ protected Response execute(OkHttpClient okHttpClient, Request request) { } } - protected void executeAndStream(Request request, ResultCallback callback, SimpleChannelInboundHandler handler) { - executeAndStream(okHttpClient, request, callback, handler); + protected void executeAndStream(Request request, ResultCallback callback, Consumer sourceConsumer) { + executeAndStream(okHttpClient, request, callback, sourceConsumer); } - protected void executeAndStream(OkHttpClient okHttpClient, Request request, ResultCallback callback, SimpleChannelInboundHandler handler) { - // TODO proper thread management - Thread thread = new Thread() { - @Override - @SneakyThrows - public void run() { - try ( - Response response = execute(okHttpClient, request.newBuilder().tag("streaming").build()); - BufferedSource source = response.body().source(); - InputStream inputStream = source.inputStream(); - ) { - AtomicBoolean shouldStop = new AtomicBoolean(); - callback.onStart(() -> { - shouldStop.set(true); - response.close(); - }); - - byte[] buffer = new byte[4 * 1024]; - while (!(shouldStop.get() || source.exhausted())) { - int bytesReceived = inputStream.read(buffer); - - int offset = 0; - for (int i = 0; i < bytesReceived; i++) { - // some handlers like JsonResponseCallbackHandler do not work with multi-line buffers - boolean isLineBreak = buffer[i] == '\n'; - boolean isEndOfBuffer = i == bytesReceived - 1; - if (isLineBreak || isEndOfBuffer) { - handler.channelRead(null, Unpooled.wrappedBuffer(buffer, offset, i - offset + 1)); - offset = i + 1; - } - } + protected void executeAndStream(OkHttpClient okHttpClient, Request request, ResultCallback callback, Consumer sourceConsumer) { + Thread thread = new Thread(DockerClientFactory.TESTCONTAINERS_THREAD_GROUP, () -> { + try ( + Response response = execute(okHttpClient, request.newBuilder().tag("streaming").build()); + BufferedSource source = response.body().source(); + ) { + callback.onStart(response); + sourceConsumer.accept(source); + callback.onComplete(); + } catch (Exception e) { + callback.onError(e); + } + }, "tc-okhttp-stream-" + Objects.hashCode(request)); + thread.setDaemon(true); + + thread.start(); + } + + @RequiredArgsConstructor + @FieldDefaults(makeFinal = true, level = AccessLevel.PRIVATE) + private static class JsonSink implements Consumer { + + private static ObjectMapper objectMapper = new ObjectMapper() + .configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false); + + TypeReference typeReference; + + ResultCallback resultCallback; + + @Override + public void accept(BufferedSource source) { + try (InputStream src = source.inputStream()) { + MappingIterator iterator = objectMapper.readerFor(typeReference).readValues(src); + while (iterator.hasNextValue() && !source.exhausted()) { + resultCallback.onNext(iterator.nextValue()); + } + } catch (Exception e) { + resultCallback.onError(e); + } + } + } + + @RequiredArgsConstructor + @FieldDefaults(makeFinal = true, level = AccessLevel.PRIVATE) + private static class FramedSink implements Consumer { + + private static final int HEADER_SIZE = 8; + + private final ResultCallback resultCallback; + + @Override + public void accept(BufferedSource source) { + try { + while (!source.exhausted()) { + // See https://docs.docker.com/engine/api/v1.37/#operation/ContainerAttach + if(!source.request(HEADER_SIZE)) { + return; + } + StreamType streamType = streamType(source.readByte()); + source.skip(3); + int payloadSize = source.readInt(); + + if(!source.request(payloadSize)) { + return; } - callback.onComplete(); - } catch (Exception e) { - callback.onError(e); + byte[] payload = source.readByteArray(payloadSize); + + resultCallback.onNext(new Frame(streamType, payload)); } + } catch (Exception e) { + resultCallback.onError(e); } - }; + } - thread.start(); + private static StreamType streamType(byte streamType) { + switch (streamType) { + case 0: + return StreamType.STDIN; + case 1: + return StreamType.STDOUT; + case 2: + return StreamType.STDERR; + default: + return StreamType.RAW; + } + } } } From 36353eecd30fc97049566637f524127d10f0afb4 Mon Sep 17 00:00:00 2001 From: Sergei Egorov Date: Fri, 14 Sep 2018 22:56:05 +0200 Subject: [PATCH 2/4] use `BufferedReader` --- .../okhttp/OkHttpInvocationBuilder.java | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/core/src/main/java/org/testcontainers/dockerclient/transport/okhttp/OkHttpInvocationBuilder.java b/core/src/main/java/org/testcontainers/dockerclient/transport/okhttp/OkHttpInvocationBuilder.java index 0fa919e3de4..b1e8f7bb2c0 100644 --- a/core/src/main/java/org/testcontainers/dockerclient/transport/okhttp/OkHttpInvocationBuilder.java +++ b/core/src/main/java/org/testcontainers/dockerclient/transport/okhttp/OkHttpInvocationBuilder.java @@ -36,9 +36,11 @@ import org.testcontainers.DockerClientFactory; import javax.annotation.Nullable; +import java.io.BufferedReader; import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; +import java.io.InputStreamReader; import java.lang.reflect.Field; import java.util.Objects; import java.util.function.Consumer; @@ -322,10 +324,17 @@ private static class JsonSink implements Consumer { @Override public void accept(BufferedSource source) { - try (InputStream src = source.inputStream()) { - MappingIterator iterator = objectMapper.readerFor(typeReference).readValues(src); - while (iterator.hasNextValue() && !source.exhausted()) { - resultCallback.onNext(iterator.nextValue()); + try ( + InputStream src = source.inputStream(); + BufferedReader reader = new BufferedReader(new InputStreamReader(src)) + ) { + while (true) { + String line = reader.readLine(); + if (line == null) { + break; + } + + resultCallback.onNext(objectMapper.readValue(line, typeReference)); } } catch (Exception e) { resultCallback.onError(e); From 7a4a35416df08115f8d0450112678adcebd07393 Mon Sep 17 00:00:00 2001 From: Sergei Egorov Date: Sat, 15 Sep 2018 15:31:16 +0200 Subject: [PATCH 3/4] simplify JsonSink --- .../transport/okhttp/OkHttpInvocationBuilder.java | 10 ++-------- 1 file changed, 2 insertions(+), 8 deletions(-) diff --git a/core/src/main/java/org/testcontainers/dockerclient/transport/okhttp/OkHttpInvocationBuilder.java b/core/src/main/java/org/testcontainers/dockerclient/transport/okhttp/OkHttpInvocationBuilder.java index b1e8f7bb2c0..1de644812e9 100644 --- a/core/src/main/java/org/testcontainers/dockerclient/transport/okhttp/OkHttpInvocationBuilder.java +++ b/core/src/main/java/org/testcontainers/dockerclient/transport/okhttp/OkHttpInvocationBuilder.java @@ -2,7 +2,6 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.core.type.TypeReference; -import com.fasterxml.jackson.databind.MappingIterator; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.SerializationFeature; import com.github.dockerjava.api.async.ResultCallback; @@ -36,11 +35,9 @@ import org.testcontainers.DockerClientFactory; import javax.annotation.Nullable; -import java.io.BufferedReader; import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; -import java.io.InputStreamReader; import java.lang.reflect.Field; import java.util.Objects; import java.util.function.Consumer; @@ -324,12 +321,9 @@ private static class JsonSink implements Consumer { @Override public void accept(BufferedSource source) { - try ( - InputStream src = source.inputStream(); - BufferedReader reader = new BufferedReader(new InputStreamReader(src)) - ) { + try { while (true) { - String line = reader.readLine(); + String line = source.readUtf8Line(); if (line == null) { break; } From f72a9e47f5cff2fdf34dd40601829b1a9c155146 Mon Sep 17 00:00:00 2001 From: Sergei Egorov Date: Sat, 15 Sep 2018 15:40:08 +0200 Subject: [PATCH 4/4] code cleanup --- .../transport/okhttp/OkHttpInvocationBuilder.java | 8 +++----- .../dockerclient/transport/okhttp/OkHttpWebTarget.java | 4 +++- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/core/src/main/java/org/testcontainers/dockerclient/transport/okhttp/OkHttpInvocationBuilder.java b/core/src/main/java/org/testcontainers/dockerclient/transport/okhttp/OkHttpInvocationBuilder.java index 1de644812e9..ee35cde2ce8 100644 --- a/core/src/main/java/org/testcontainers/dockerclient/transport/okhttp/OkHttpInvocationBuilder.java +++ b/core/src/main/java/org/testcontainers/dockerclient/transport/okhttp/OkHttpInvocationBuilder.java @@ -3,7 +3,6 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; -import com.fasterxml.jackson.databind.SerializationFeature; import com.github.dockerjava.api.async.ResultCallback; import com.github.dockerjava.api.exception.BadRequestException; import com.github.dockerjava.api.exception.ConflictException; @@ -200,7 +199,7 @@ public void post(TypeReference typeReference, ResultCallback resultCal executeAndStream( request, resultCallback, - new JsonSink<>(typeReference, resultCallback) + new JsonSink<>(objectMapper, typeReference, resultCallback) ); } @@ -312,8 +311,7 @@ protected void executeAndStream(OkHttpClient okHttpClient, Request request, @FieldDefaults(makeFinal = true, level = AccessLevel.PRIVATE) private static class JsonSink implements Consumer { - private static ObjectMapper objectMapper = new ObjectMapper() - .configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false); + ObjectMapper objectMapper; TypeReference typeReference; @@ -342,7 +340,7 @@ private static class FramedSink implements Consumer { private static final int HEADER_SIZE = 8; - private final ResultCallback resultCallback; + ResultCallback resultCallback; @Override public void accept(BufferedSource source) { diff --git a/core/src/main/java/org/testcontainers/dockerclient/transport/okhttp/OkHttpWebTarget.java b/core/src/main/java/org/testcontainers/dockerclient/transport/okhttp/OkHttpWebTarget.java index 85fe8c22288..7e12b0b4541 100644 --- a/core/src/main/java/org/testcontainers/dockerclient/transport/okhttp/OkHttpWebTarget.java +++ b/core/src/main/java/org/testcontainers/dockerclient/transport/okhttp/OkHttpWebTarget.java @@ -2,6 +2,7 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializationFeature; import com.github.dockerjava.core.InvocationBuilder; import com.github.dockerjava.core.WebTarget; import com.google.common.collect.HashMultimap; @@ -24,7 +25,8 @@ @Value class OkHttpWebTarget implements WebTarget { - private static final ObjectMapper MAPPER = new ObjectMapper(); + private static final ObjectMapper MAPPER = new ObjectMapper() + .disable(SerializationFeature.FAIL_ON_EMPTY_BEANS); OkHttpClient okHttpClient;