|
18 | 18 |
|
19 | 19 | import java.io.ByteArrayOutputStream;
|
20 | 20 | import java.io.IOException;
|
21 |
| -import java.io.UncheckedIOException; |
22 |
| -import java.util.Collections; |
| 21 | +import java.net.URI; |
23 | 22 | import java.util.List;
|
24 | 23 | import java.util.zip.GZIPOutputStream;
|
25 | 24 |
|
26 |
| -import zipkin2.Call; |
27 |
| -import zipkin2.CheckResult; |
28 |
| -import zipkin2.codec.Encoding; |
29 |
| -import zipkin2.reporter.BytesMessageEncoder; |
30 |
| -import zipkin2.reporter.ClosedSenderException; |
31 |
| -import zipkin2.reporter.Sender; |
| 25 | +import zipkin2.reporter.BaseHttpSender; |
| 26 | +import zipkin2.reporter.BytesMessageSender; |
| 27 | +import zipkin2.reporter.Encoding; |
| 28 | +import zipkin2.reporter.HttpEndpointSupplier.Factory; |
32 | 29 |
|
33 | 30 | import org.springframework.http.HttpHeaders;
|
34 | 31 | import org.springframework.util.unit.DataSize;
|
35 | 32 |
|
36 | 33 | /**
|
37 |
| - * A Zipkin {@link Sender} that uses an HTTP client to send JSON spans. Supports automatic |
38 |
| - * compression with gzip. |
| 34 | + * A Zipkin {@link BytesMessageSender} that uses an HTTP client to send JSON spans. |
| 35 | + * Supports automatic compression with gzip. |
39 | 36 | *
|
40 | 37 | * @author Moritz Halbritter
|
41 | 38 | * @author Stefan Bratanov
|
42 | 39 | */
|
43 |
| -abstract class HttpSender extends Sender { |
| 40 | +abstract class HttpSender extends BaseHttpSender<URI, byte[]> { |
44 | 41 |
|
45 |
| - private static final DataSize MESSAGE_MAX_SIZE = DataSize.ofKilobytes(512); |
46 |
| - |
47 |
| - private volatile boolean closed; |
48 |
| - |
49 |
| - @Override |
50 |
| - public Encoding encoding() { |
51 |
| - return Encoding.JSON; |
52 |
| - } |
| 42 | + /** |
| 43 | + * Only use gzip compression on data which is bigger than this in bytes. |
| 44 | + */ |
| 45 | + private static final DataSize COMPRESSION_THRESHOLD = DataSize.ofKilobytes(1); |
53 | 46 |
|
54 |
| - @Override |
55 |
| - public int messageMaxBytes() { |
56 |
| - return (int) MESSAGE_MAX_SIZE.toBytes(); |
| 47 | + HttpSender(Encoding encoding, Factory endpointSupplierFactory, String endpoint) { |
| 48 | + super(encoding, endpointSupplierFactory, endpoint); |
57 | 49 | }
|
58 | 50 |
|
59 | 51 | @Override
|
60 |
| - public int messageSizeInBytes(List<byte[]> encodedSpans) { |
61 |
| - return encoding().listSizeInBytes(encodedSpans); |
| 52 | + protected URI newEndpoint(String endpoint) { |
| 53 | + return URI.create(endpoint); |
62 | 54 | }
|
63 | 55 |
|
64 | 56 | @Override
|
65 |
| - public int messageSizeInBytes(int encodedSizeInBytes) { |
66 |
| - return encoding().listSizeInBytes(encodedSizeInBytes); |
| 57 | + protected byte[] newBody(List<byte[]> list) { |
| 58 | + return this.encoding.encode(list); |
67 | 59 | }
|
68 | 60 |
|
69 | 61 | @Override
|
70 |
| - public CheckResult check() { |
71 |
| - try { |
72 |
| - sendSpans(Collections.emptyList()).execute(); |
73 |
| - return CheckResult.OK; |
| 62 | + protected void postSpans(URI endpoint, byte[] body) throws IOException { |
| 63 | + HttpHeaders headers = getDefaultHeaders(); |
| 64 | + if (needsCompression(body)) { |
| 65 | + body = compress(body); |
| 66 | + headers.set("Content-Encoding", "gzip"); |
74 | 67 | }
|
75 |
| - catch (IOException | RuntimeException ex) { |
76 |
| - return CheckResult.failed(ex); |
77 |
| - } |
78 |
| - } |
79 |
| - |
80 |
| - @Override |
81 |
| - public void close() throws IOException { |
82 |
| - this.closed = true; |
| 68 | + postSpans(endpoint, headers, body); |
83 | 69 | }
|
84 | 70 |
|
85 | 71 | /**
|
86 |
| - * The returned {@link HttpPostCall} will send span(s) as a POST to a zipkin endpoint |
87 |
| - * when executed. |
88 |
| - * @param batchedEncodedSpans list of encoded spans as a byte array |
89 |
| - * @return an instance of a Zipkin {@link Call} which can be executed |
| 72 | + * This will send span(s) as a POST to a zipkin endpoint. |
| 73 | + * @param endpoint the POST endpoint. For example, http://localhost:9411/api/v2/spans. |
| 74 | + * @param headers headers for the POST request |
| 75 | + * @param body list of possibly gzipped, encoded spans. |
90 | 76 | */
|
91 |
| - protected abstract HttpPostCall sendSpans(byte[] batchedEncodedSpans); |
| 77 | + abstract void postSpans(URI endpoint, HttpHeaders headers, byte[] body); |
92 | 78 |
|
93 |
| - @Override |
94 |
| - public Call<Void> sendSpans(List<byte[]> encodedSpans) { |
95 |
| - if (this.closed) { |
96 |
| - throw new ClosedSenderException(); |
97 |
| - } |
98 |
| - return sendSpans(BytesMessageEncoder.JSON.encode(encodedSpans)); |
| 79 | + HttpHeaders getDefaultHeaders() { |
| 80 | + HttpHeaders headers = new HttpHeaders(); |
| 81 | + headers.set("b3", "0"); |
| 82 | + headers.set("Content-Type", this.encoding.mediaType()); |
| 83 | + return headers; |
99 | 84 | }
|
100 | 85 |
|
101 |
| - abstract static class HttpPostCall extends Call.Base<Void> { |
102 |
| - |
103 |
| - /** |
104 |
| - * Only use gzip compression on data which is bigger than this in bytes. |
105 |
| - */ |
106 |
| - private static final DataSize COMPRESSION_THRESHOLD = DataSize.ofKilobytes(1); |
107 |
| - |
108 |
| - private final byte[] body; |
109 |
| - |
110 |
| - HttpPostCall(byte[] body) { |
111 |
| - this.body = body; |
112 |
| - } |
113 |
| - |
114 |
| - protected byte[] getBody() { |
115 |
| - if (needsCompression()) { |
116 |
| - return compress(this.body); |
117 |
| - } |
118 |
| - return this.body; |
119 |
| - } |
120 |
| - |
121 |
| - protected byte[] getUncompressedBody() { |
122 |
| - return this.body; |
123 |
| - } |
124 |
| - |
125 |
| - protected HttpHeaders getDefaultHeaders() { |
126 |
| - HttpHeaders headers = new HttpHeaders(); |
127 |
| - headers.set("b3", "0"); |
128 |
| - headers.set("Content-Type", "application/json"); |
129 |
| - if (needsCompression()) { |
130 |
| - headers.set("Content-Encoding", "gzip"); |
131 |
| - } |
132 |
| - return headers; |
133 |
| - } |
134 |
| - |
135 |
| - private boolean needsCompression() { |
136 |
| - return this.body.length > COMPRESSION_THRESHOLD.toBytes(); |
137 |
| - } |
| 86 | + private boolean needsCompression(byte[] body) { |
| 87 | + return body.length > COMPRESSION_THRESHOLD.toBytes(); |
| 88 | + } |
138 | 89 |
|
139 |
| - private byte[] compress(byte[] input) { |
140 |
| - ByteArrayOutputStream result = new ByteArrayOutputStream(); |
141 |
| - try (GZIPOutputStream gzip = new GZIPOutputStream(result)) { |
142 |
| - gzip.write(input); |
143 |
| - } |
144 |
| - catch (IOException ex) { |
145 |
| - throw new UncheckedIOException(ex); |
146 |
| - } |
147 |
| - return result.toByteArray(); |
| 90 | + private byte[] compress(byte[] input) throws IOException { |
| 91 | + ByteArrayOutputStream result = new ByteArrayOutputStream(); |
| 92 | + try (GZIPOutputStream gzip = new GZIPOutputStream(result)) { |
| 93 | + gzip.write(input); |
148 | 94 | }
|
149 |
| - |
| 95 | + return result.toByteArray(); |
150 | 96 | }
|
151 | 97 |
|
152 | 98 | }
|
0 commit comments