Skip to content

Commit 8c27b67

Browse files
slandelleslandelle
committed
format
1 parent 931bb1d commit 8c27b67

File tree

1 file changed

+93
-92
lines changed

1 file changed

+93
-92
lines changed

‎client/src/main/java/org/asynchttpclient/netty/request/body/NettyReactiveStreamsBody.java

+93-92
Original file line numberDiff line numberDiff line change
@@ -32,96 +32,97 @@
3232

3333
public class NettyReactiveStreamsBody implements NettyBody {
3434

35-
private static final Logger LOGGER = LoggerFactory.getLogger(NettyReactiveStreamsBody.class);
36-
private static final String NAME_IN_CHANNEL_PIPELINE = "request-body-streamer";
37-
38-
private final Publisher<ByteBuf> publisher;
39-
40-
private final long contentLength;
41-
42-
public NettyReactiveStreamsBody(Publisher<ByteBuf> publisher, long contentLength) {
43-
this.publisher = publisher;
44-
this.contentLength = contentLength;
45-
}
46-
47-
@Override
48-
public long getContentLength() {
49-
return contentLength;
50-
}
51-
52-
@Override
53-
public void write(Channel channel, NettyResponseFuture<?> future) throws IOException {
54-
if (future.isStreamConsumed()) {
55-
LOGGER.warn("Stream has already been consumed and cannot be reset");
56-
} else {
57-
future.setStreamConsumed(true);
58-
NettySubscriber subscriber = new NettySubscriber(channel, future);
59-
channel.pipeline().addLast(NAME_IN_CHANNEL_PIPELINE, subscriber);
60-
publisher.subscribe(new SubscriberAdapter(subscriber));
61-
}
62-
}
63-
64-
private static class SubscriberAdapter implements Subscriber<ByteBuf> {
65-
private final Subscriber<HttpContent> subscriber;
66-
67-
public SubscriberAdapter(Subscriber<HttpContent> subscriber) {
68-
this.subscriber = subscriber;
69-
}
70-
71-
@Override
72-
public void onSubscribe(Subscription s) {
73-
subscriber.onSubscribe(s);
74-
}
75-
76-
@Override
77-
public void onNext(ByteBuf buffer) {
78-
HttpContent content = new DefaultHttpContent(buffer);
79-
subscriber.onNext(content);
80-
}
81-
82-
@Override
83-
public void onError(Throwable t) {
84-
subscriber.onError(t);
85-
}
86-
87-
@Override
88-
public void onComplete() {
89-
subscriber.onComplete();
90-
}
91-
}
92-
93-
private static class NettySubscriber extends HandlerSubscriber<HttpContent> {
94-
private static final Logger LOGGER = LoggerFactory.getLogger(NettySubscriber.class);
95-
96-
private final Channel channel;
97-
private final NettyResponseFuture<?> future;
98-
99-
public NettySubscriber(Channel channel, NettyResponseFuture<?> future) {
100-
super(channel.eventLoop());
101-
this.channel = channel;
102-
this.future = future;
103-
}
104-
105-
@Override
106-
protected void complete() {
107-
channel.eventLoop().execute(() -> channel.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT).addListener(future -> removeFromPipeline()));
108-
}
109-
110-
@Override
111-
protected void error(Throwable error) {
112-
if (error == null)
113-
throw null;
114-
removeFromPipeline();
115-
future.abort(error);
116-
}
117-
118-
private void removeFromPipeline() {
119-
try {
120-
channel.pipeline().remove(this);
121-
LOGGER.debug(String.format("Removed handler %s from pipeline.", NAME_IN_CHANNEL_PIPELINE));
122-
} catch (NoSuchElementException e) {
123-
LOGGER.debug(String.format("Failed to remove handler %s from pipeline.", NAME_IN_CHANNEL_PIPELINE), e);
124-
}
125-
}
126-
}
35+
private static final Logger LOGGER = LoggerFactory.getLogger(NettyReactiveStreamsBody.class);
36+
private static final String NAME_IN_CHANNEL_PIPELINE = "request-body-streamer";
37+
38+
private final Publisher<ByteBuf> publisher;
39+
40+
private final long contentLength;
41+
42+
public NettyReactiveStreamsBody(Publisher<ByteBuf> publisher, long contentLength) {
43+
this.publisher = publisher;
44+
this.contentLength = contentLength;
45+
}
46+
47+
@Override
48+
public long getContentLength() {
49+
return contentLength;
50+
}
51+
52+
@Override
53+
public void write(Channel channel, NettyResponseFuture<?> future) throws IOException {
54+
if (future.isStreamConsumed()) {
55+
LOGGER.warn("Stream has already been consumed and cannot be reset");
56+
} else {
57+
future.setStreamConsumed(true);
58+
NettySubscriber subscriber = new NettySubscriber(channel, future);
59+
channel.pipeline().addLast(NAME_IN_CHANNEL_PIPELINE, subscriber);
60+
publisher.subscribe(new SubscriberAdapter(subscriber));
61+
}
62+
}
63+
64+
private static class SubscriberAdapter implements Subscriber<ByteBuf> {
65+
private final Subscriber<HttpContent> subscriber;
66+
67+
public SubscriberAdapter(Subscriber<HttpContent> subscriber) {
68+
this.subscriber = subscriber;
69+
}
70+
71+
@Override
72+
public void onSubscribe(Subscription s) {
73+
subscriber.onSubscribe(s);
74+
}
75+
76+
@Override
77+
public void onNext(ByteBuf buffer) {
78+
HttpContent content = new DefaultHttpContent(buffer);
79+
subscriber.onNext(content);
80+
}
81+
82+
@Override
83+
public void onError(Throwable t) {
84+
subscriber.onError(t);
85+
}
86+
87+
@Override
88+
public void onComplete() {
89+
subscriber.onComplete();
90+
}
91+
}
92+
93+
private static class NettySubscriber extends HandlerSubscriber<HttpContent> {
94+
private static final Logger LOGGER = LoggerFactory.getLogger(NettySubscriber.class);
95+
96+
private final Channel channel;
97+
private final NettyResponseFuture<?> future;
98+
99+
public NettySubscriber(Channel channel, NettyResponseFuture<?> future) {
100+
super(channel.eventLoop());
101+
this.channel = channel;
102+
this.future = future;
103+
}
104+
105+
@Override
106+
protected void complete() {
107+
channel.eventLoop().execute(() -> channel.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT)
108+
.addListener(future -> removeFromPipeline()));
109+
}
110+
111+
@Override
112+
protected void error(Throwable error) {
113+
if (error == null)
114+
throw null;
115+
removeFromPipeline();
116+
future.abort(error);
117+
}
118+
119+
private void removeFromPipeline() {
120+
try {
121+
channel.pipeline().remove(this);
122+
LOGGER.debug(String.format("Removed handler %s from pipeline.", NAME_IN_CHANNEL_PIPELINE));
123+
} catch (NoSuchElementException e) {
124+
LOGGER.debug(String.format("Failed to remove handler %s from pipeline.", NAME_IN_CHANNEL_PIPELINE), e);
125+
}
126+
}
127+
}
127128
}

0 commit comments

Comments
 (0)