Skip to content

Commit 32f313a

Browse files
slandelleslandelle
committed
Drop AdditionalChannelInitializer for a Consumer
1 parent 391b4d8 commit 32f313a

File tree

5 files changed

+163
-154
lines changed

5 files changed

+163
-154
lines changed

‎client/src/main/java/org/asynchttpclient/AsyncCompletionHandler.java

Lines changed: 88 additions & 72 deletions
Original file line numberDiff line numberDiff line change
@@ -23,90 +23,106 @@
2323
import org.slf4j.LoggerFactory;
2424

2525
/**
26-
* An {@link AsyncHandler} augmented with an {@link #onCompleted(Response)} convenience method which gets called when the {@link Response} processing is finished. This class also
27-
* implement the {@link ProgressAsyncHandler} callback, all doing nothing except returning {@link org.asynchttpclient.AsyncHandler.State#CONTINUE}
26+
* An {@link AsyncHandler} augmented with an {@link #onCompleted(Response)}
27+
* convenience method which gets called when the {@link Response} processing is
28+
* finished. This class also implement the {@link ProgressAsyncHandler}
29+
* callback, all doing nothing except returning
30+
* {@link org.asynchttpclient.AsyncHandler.State#CONTINUE}
2831
*
29-
* @param <T> Type of the value that will be returned by the associated {@link java.util.concurrent.Future}
32+
* @param <T>
33+
* Type of the value that will be returned by the associated
34+
* {@link java.util.concurrent.Future}
3035
*/
3136
public abstract class AsyncCompletionHandler<T> implements AsyncHandler<T>, ProgressAsyncHandler<T> {
3237

33-
private static final Logger LOGGER = LoggerFactory.getLogger(AsyncCompletionHandler.class);
34-
private final Response.ResponseBuilder builder = new Response.ResponseBuilder();
38+
private static final Logger LOGGER = LoggerFactory.getLogger(AsyncCompletionHandler.class);
39+
private final Response.ResponseBuilder builder = new Response.ResponseBuilder();
3540

36-
@Override
37-
public State onBodyPartReceived(HttpResponseBodyPart content) throws Exception {
38-
builder.accumulate(content);
39-
return State.CONTINUE;
40-
}
41+
@Override
42+
public State onStatusReceived(HttpResponseStatus status) throws Exception {
43+
builder.reset();
44+
builder.accumulate(status);
45+
return State.CONTINUE;
46+
}
4147

42-
@Override
43-
public State onStatusReceived(HttpResponseStatus status) throws Exception {
44-
builder.reset();
45-
builder.accumulate(status);
46-
return State.CONTINUE;
47-
}
48+
@Override
49+
public State onHeadersReceived(HttpHeaders headers) throws Exception {
50+
builder.accumulate(headers);
51+
return State.CONTINUE;
52+
}
4853

49-
@Override
50-
public State onHeadersReceived(HttpHeaders headers) throws Exception {
51-
builder.accumulate(headers);
52-
return State.CONTINUE;
53-
}
54+
@Override
55+
public State onBodyPartReceived(HttpResponseBodyPart content) throws Exception {
56+
builder.accumulate(content);
57+
return State.CONTINUE;
58+
}
5459

55-
@Override
56-
public State onTrailingHeadersReceived(HttpHeaders headers) throws Exception {
57-
builder.accumulate(headers);
58-
return State.CONTINUE;
59-
}
60+
@Override
61+
public State onTrailingHeadersReceived(HttpHeaders headers) throws Exception {
62+
builder.accumulate(headers);
63+
return State.CONTINUE;
64+
}
6065

61-
@Override
62-
public final T onCompleted() throws Exception {
63-
return onCompleted(builder.build());
64-
}
66+
@Override
67+
public final T onCompleted() throws Exception {
68+
return onCompleted(builder.build());
69+
}
6570

66-
@Override
67-
public void onThrowable(Throwable t) {
68-
LOGGER.debug(t.getMessage(), t);
69-
}
71+
@Override
72+
public void onThrowable(Throwable t) {
73+
LOGGER.debug(t.getMessage(), t);
74+
}
7075

71-
/**
72-
* Invoked once the HTTP response processing is finished.
73-
*
74-
* @param response The {@link Response}
75-
* @return T Value that will be returned by the associated {@link java.util.concurrent.Future}
76-
* @throws Exception if something wrong happens
77-
*/
78-
abstract public T onCompleted(Response response) throws Exception;
76+
/**
77+
* Invoked once the HTTP response processing is finished.
78+
*
79+
* @param response
80+
* The {@link Response}
81+
* @return T Value that will be returned by the associated
82+
* {@link java.util.concurrent.Future}
83+
* @throws Exception
84+
* if something wrong happens
85+
*/
86+
abstract public T onCompleted(Response response) throws Exception;
7987

80-
/**
81-
* Invoked when the HTTP headers have been fully written on the I/O socket.
82-
*
83-
* @return a {@link org.asynchttpclient.AsyncHandler.State} telling to CONTINUE or ABORT the current processing.
84-
*/
85-
@Override
86-
public State onHeadersWritten() {
87-
return State.CONTINUE;
88-
}
88+
/**
89+
* Invoked when the HTTP headers have been fully written on the I/O socket.
90+
*
91+
* @return a {@link org.asynchttpclient.AsyncHandler.State} telling to CONTINUE
92+
* or ABORT the current processing.
93+
*/
94+
@Override
95+
public State onHeadersWritten() {
96+
return State.CONTINUE;
97+
}
8998

90-
/**
91-
* Invoked when the content (a {@link java.io.File}, {@link String} or {@link java.io.InputStream} has been fully written on the I/O socket.
92-
*
93-
* @return a {@link org.asynchttpclient.AsyncHandler.State} telling to CONTINUE or ABORT the current processing.
94-
*/
95-
@Override
96-
public State onContentWritten() {
97-
return State.CONTINUE;
98-
}
99+
/**
100+
* Invoked when the content (a {@link java.io.File}, {@link String} or
101+
* {@link java.io.InputStream} has been fully written on the I/O socket.
102+
*
103+
* @return a {@link org.asynchttpclient.AsyncHandler.State} telling to CONTINUE
104+
* or ABORT the current processing.
105+
*/
106+
@Override
107+
public State onContentWritten() {
108+
return State.CONTINUE;
109+
}
99110

100-
/**
101-
* Invoked when the I/O operation associated with the {@link Request} body as been progressed.
102-
*
103-
* @param amount The amount of bytes to transfer
104-
* @param current The amount of bytes transferred
105-
* @param total The total number of bytes transferred
106-
* @return a {@link org.asynchttpclient.AsyncHandler.State} telling to CONTINUE or ABORT the current processing.
107-
*/
108-
@Override
109-
public State onContentWriteProgress(long amount, long current, long total) {
110-
return State.CONTINUE;
111-
}
111+
/**
112+
* Invoked when the I/O operation associated with the {@link Request} body as
113+
* been progressed.
114+
*
115+
* @param amount
116+
* The amount of bytes to transfer
117+
* @param current
118+
* The amount of bytes transferred
119+
* @param total
120+
* The total number of bytes transferred
121+
* @return a {@link org.asynchttpclient.AsyncHandler.State} telling to CONTINUE
122+
* or ABORT the current processing.
123+
*/
124+
@Override
125+
public State onContentWriteProgress(long amount, long current, long total) {
126+
return State.CONTINUE;
127+
}
112128
}

‎client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java

Lines changed: 11 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -13,17 +13,10 @@
1313
*/
1414
package org.asynchttpclient;
1515

16-
import io.netty.buffer.ByteBuf;
17-
import io.netty.buffer.ByteBufAllocator;
18-
import io.netty.channel.Channel;
19-
import io.netty.channel.ChannelOption;
20-
import io.netty.channel.EventLoopGroup;
21-
import io.netty.handler.ssl.SslContext;
22-
import io.netty.util.Timer;
23-
2416
import java.util.List;
2517
import java.util.Map;
2618
import java.util.concurrent.ThreadFactory;
19+
import java.util.function.Consumer;
2720

2821
import org.asynchttpclient.channel.ChannelPool;
2922
import org.asynchttpclient.channel.KeepAliveStrategy;
@@ -35,6 +28,14 @@
3528
import org.asynchttpclient.proxy.ProxyServer;
3629
import org.asynchttpclient.proxy.ProxyServerSelector;
3730

31+
import io.netty.buffer.ByteBuf;
32+
import io.netty.buffer.ByteBufAllocator;
33+
import io.netty.channel.Channel;
34+
import io.netty.channel.ChannelOption;
35+
import io.netty.channel.EventLoopGroup;
36+
import io.netty.handler.ssl.SslContext;
37+
import io.netty.util.Timer;
38+
3839
public interface AsyncHttpClientConfig {
3940

4041
/**
@@ -272,9 +273,9 @@ public interface AsyncHttpClientConfig {
272273

273274
boolean isUseNativeTransport();
274275

275-
AdditionalChannelInitializer getHttpAdditionalChannelInitializer();
276+
Consumer<Channel> getHttpAdditionalChannelInitializer();
276277

277-
AdditionalChannelInitializer getWsAdditionalChannelInitializer();
278+
Consumer<Channel> getWsAdditionalChannelInitializer();
278279

279280
ResponseBodyPartFactory getResponseBodyPartFactory();
280281

@@ -302,11 +303,6 @@ public interface AsyncHttpClientConfig {
302303

303304
int getIoThreadsCount();
304305

305-
interface AdditionalChannelInitializer {
306-
307-
void initChannel(Channel channel) throws Exception;
308-
}
309-
310306
enum ResponseBodyPartFactory {
311307

312308
EAGER {

‎client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java

Lines changed: 19 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -16,21 +16,12 @@
1616
package org.asynchttpclient;
1717

1818
import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.*;
19-
import io.netty.buffer.ByteBufAllocator;
20-
import io.netty.channel.ChannelOption;
21-
import io.netty.channel.EventLoopGroup;
22-
import io.netty.handler.ssl.SslContext;
23-
import io.netty.util.Timer;
2419

2520
import java.io.IOException;
2621
import java.io.InputStream;
27-
import java.util.Collections;
28-
import java.util.HashMap;
29-
import java.util.LinkedList;
30-
import java.util.List;
31-
import java.util.Map;
32-
import java.util.Properties;
22+
import java.util.*;
3323
import java.util.concurrent.ThreadFactory;
24+
import java.util.function.Consumer;
3425

3526
import org.asynchttpclient.channel.ChannelPool;
3627
import org.asynchttpclient.channel.DefaultKeepAliveStrategy;
@@ -42,6 +33,13 @@
4233
import org.asynchttpclient.proxy.ProxyServerSelector;
4334
import org.asynchttpclient.util.ProxyUtils;
4435

36+
import io.netty.buffer.ByteBufAllocator;
37+
import io.netty.channel.Channel;
38+
import io.netty.channel.ChannelOption;
39+
import io.netty.channel.EventLoopGroup;
40+
import io.netty.handler.ssl.SslContext;
41+
import io.netty.util.Timer;
42+
4543
/**
4644
* Configuration class to use with a {@link AsyncHttpClient}. System property can be also used to configure this object default behavior by doing: <br>
4745
* -Dorg.asynchttpclient.nameOfTheProperty
@@ -132,8 +130,8 @@ public class DefaultAsyncHttpClientConfig implements AsyncHttpClientConfig {
132130
private final int soRcvBuf;
133131
private final Timer nettyTimer;
134132
private final ThreadFactory threadFactory;
135-
private final AdditionalChannelInitializer httpAdditionalChannelInitializer;
136-
private final AdditionalChannelInitializer wsAdditionalChannelInitializer;
133+
private final Consumer<Channel> httpAdditionalChannelInitializer;
134+
private final Consumer<Channel> wsAdditionalChannelInitializer;
137135
private final ResponseBodyPartFactory responseBodyPartFactory;
138136
private final int ioThreadsCount;
139137

@@ -210,8 +208,8 @@ private DefaultAsyncHttpClientConfig(//
210208
ByteBufAllocator allocator,//
211209
Timer nettyTimer,//
212210
ThreadFactory threadFactory,//
213-
AdditionalChannelInitializer httpAdditionalChannelInitializer,//
214-
AdditionalChannelInitializer wsAdditionalChannelInitializer,//
211+
Consumer<Channel> httpAdditionalChannelInitializer,//
212+
Consumer<Channel> wsAdditionalChannelInitializer,//
215213
ResponseBodyPartFactory responseBodyPartFactory,//
216214
int ioThreadsCount) {
217215

@@ -602,12 +600,12 @@ public ThreadFactory getThreadFactory() {
602600
}
603601

604602
@Override
605-
public AdditionalChannelInitializer getHttpAdditionalChannelInitializer() {
603+
public Consumer<Channel> getHttpAdditionalChannelInitializer() {
606604
return httpAdditionalChannelInitializer;
607605
}
608606

609607
@Override
610-
public AdditionalChannelInitializer getWsAdditionalChannelInitializer() {
608+
public Consumer<Channel> getWsAdditionalChannelInitializer() {
611609
return wsAdditionalChannelInitializer;
612610
}
613611

@@ -700,8 +698,8 @@ public static class Builder {
700698
private EventLoopGroup eventLoopGroup;
701699
private Timer nettyTimer;
702700
private ThreadFactory threadFactory;
703-
private AdditionalChannelInitializer httpAdditionalChannelInitializer;
704-
private AdditionalChannelInitializer wsAdditionalChannelInitializer;
701+
private Consumer<Channel> httpAdditionalChannelInitializer;
702+
private Consumer<Channel> wsAdditionalChannelInitializer;
705703
private ResponseBodyPartFactory responseBodyPartFactory = ResponseBodyPartFactory.EAGER;
706704
private int ioThreadsCount = defaultIoThreadsCount();
707705

@@ -1117,12 +1115,12 @@ public Builder setThreadFactory(ThreadFactory threadFactory) {
11171115
return this;
11181116
}
11191117

1120-
public Builder setHttpAdditionalChannelInitializer(AdditionalChannelInitializer httpAdditionalChannelInitializer) {
1118+
public Builder setHttpAdditionalChannelInitializer(Consumer<Channel> httpAdditionalChannelInitializer) {
11211119
this.httpAdditionalChannelInitializer = httpAdditionalChannelInitializer;
11221120
return this;
11231121
}
11241122

1125-
public Builder setWsAdditionalChannelInitializer(AdditionalChannelInitializer wsAdditionalChannelInitializer) {
1123+
public Builder setWsAdditionalChannelInitializer(Consumer<Channel> wsAdditionalChannelInitializer) {
11261124
this.wsAdditionalChannelInitializer = wsAdditionalChannelInitializer;
11271125
return this;
11281126
}

‎client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -232,7 +232,7 @@ protected void initChannel(Channel ch) throws Exception {
232232
}
233233

234234
if (config.getHttpAdditionalChannelInitializer() != null)
235-
config.getHttpAdditionalChannelInitializer().initChannel(ch);
235+
config.getHttpAdditionalChannelInitializer().accept(ch);
236236
}
237237
});
238238

@@ -249,7 +249,7 @@ protected void initChannel(Channel ch) throws Exception {
249249
}
250250

251251
if (config.getWsAdditionalChannelInitializer() != null)
252-
config.getWsAdditionalChannelInitializer().initChannel(ch);
252+
config.getWsAdditionalChannelInitializer().accept(ch);
253253
}
254254
});
255255
}

0 commit comments

Comments
 (0)