Skip to content

Commit d8a0355

Browse files
author
Stephane Landelle
committed
minor clean up
1 parent 82cb11a commit d8a0355

File tree

1 file changed

+39
-50
lines changed

1 file changed

+39
-50
lines changed

‎providers/netty/src/main/java/org/asynchttpclient/providers/netty/channel/DefaultChannelPool.java

Lines changed: 39 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@
3838
*/
3939
public class DefaultChannelPool implements ChannelPool {
4040

41-
private final static Logger log = LoggerFactory.getLogger(DefaultChannelPool.class);
41+
private final static Logger LOGGER = LoggerFactory.getLogger(DefaultChannelPool.class);
4242
private final ConcurrentHashMapV8<String, ConcurrentLinkedQueue<IdleChannel>> connectionsPool = new ConcurrentHashMapV8<String, ConcurrentLinkedQueue<IdleChannel>>();
4343
private final ConcurrentHashMapV8<Channel, IdleChannel> channel2IdleChannel = new ConcurrentHashMapV8<Channel, IdleChannel>();
4444
private final ConcurrentHashMapV8<Channel, Long> channel2CreationDate = new ConcurrentHashMapV8<Channel, Long>();
@@ -81,30 +81,24 @@ private void scheduleNewIdleChannelDetector(TimerTask task) {
8181
nettyTimer.newTimeout(task, maxIdleTime, TimeUnit.MILLISECONDS);
8282
}
8383

84-
private static class IdleChannel {
85-
final String uri;
84+
private static final class IdleChannel {
85+
final String key;
8686
final Channel channel;
8787
final long start;
8888

89-
IdleChannel(String uri, Channel channel) {
90-
this.uri = uri;
89+
IdleChannel(String key, Channel channel) {
90+
if (key == null)
91+
throw new NullPointerException("key");
92+
if (channel == null)
93+
throw new NullPointerException("channel");
94+
this.key = key;
9195
this.channel = channel;
9296
this.start = millisTime();
9397
}
9498

9599
@Override
96100
public boolean equals(Object o) {
97-
if (this == o)
98-
return true;
99-
if (!(o instanceof IdleChannel))
100-
return false;
101-
102-
IdleChannel that = (IdleChannel) o;
103-
104-
if (channel != null ? !channel.equals(that.channel) : that.channel != null)
105-
return false;
106-
107-
return true;
101+
return this == o || (o instanceof IdleChannel && channel.equals(IdleChannel.class.cast(o).channel));
108102
}
109103

110104
@Override
@@ -121,11 +115,11 @@ public void run(Timeout timeout) throws Exception {
121115
if (closed.get())
122116
return;
123117

124-
if (log.isDebugEnabled()) {
118+
if (LOGGER.isDebugEnabled()) {
125119
Set<String> keys = connectionsPool.keySet();
126120

127121
for (String s : keys) {
128-
log.debug("Entry count for : {} : {}", s, connectionsPool.get(s).size());
122+
LOGGER.debug("Entry count for : {} : {}", s, connectionsPool.get(s).size());
129123
}
130124
}
131125

@@ -136,7 +130,7 @@ public void run(Timeout timeout) throws Exception {
136130
long age = currentTime - idleChannel.start;
137131
if (age > maxIdleTime) {
138132

139-
log.debug("Adding Candidate Idle Channel {}", idleChannel.channel);
133+
LOGGER.debug("Adding Candidate Idle Channel {}", idleChannel.channel);
140134

141135
// store in an unsynchronized list to minimize the impact on the ConcurrentHashMap.
142136
channelsInTimeout.add(idleChannel);
@@ -151,28 +145,28 @@ public void run(Timeout timeout) throws Exception {
151145
NettyResponseFuture<?> future = (NettyResponseFuture<?>) attachment;
152146

153147
if (!future.isDone() && !future.isCancelled()) {
154-
log.debug("Future not in appropriate state %s\n", future);
148+
LOGGER.debug("Future not in appropriate state %s\n", future);
155149
continue;
156150
}
157151
}
158152
}
159153

160154
if (remove(idleChannel)) {
161-
log.debug("Closing Idle Channel {}", idleChannel.channel);
155+
LOGGER.debug("Closing Idle Channel {}", idleChannel.channel);
162156
close(idleChannel.channel);
163157
}
164158
}
165159

166-
if (log.isTraceEnabled()) {
160+
if (LOGGER.isTraceEnabled()) {
167161
int openChannels = 0;
168162
for (ConcurrentLinkedQueue<IdleChannel> hostChannels : connectionsPool.values()) {
169163
openChannels += hostChannels.size();
170164
}
171-
log.trace(String.format("%d channel open, %d idle channels closed (times: 1st-loop=%d, 2nd-loop=%d).\n", openChannels,
165+
LOGGER.trace(String.format("%d channel open, %d idle channels closed (times: 1st-loop=%d, 2nd-loop=%d).\n", openChannels,
172166
channelsInTimeout.size(), endConcurrentLoop - currentTime, millisTime() - endConcurrentLoop));
173167
}
174168
} catch (Throwable t) {
175-
log.error("uncaught exception!", t);
169+
LOGGER.error("uncaught exception!", t);
176170
}
177171

178172
scheduleNewIdleChannelDetector(timeout.task());
@@ -183,46 +177,42 @@ public void run(Timeout timeout) throws Exception {
183177
* {@inheritDoc}
184178
*/
185179
public boolean offer(String uri, Channel channel) {
186-
if (closed.get())
180+
if (closed.get() || (!sslConnectionPoolEnabled && uri.startsWith("https")))
187181
return false;
188182

189-
if (!sslConnectionPoolEnabled && uri.startsWith("https")) {
190-
return false;
191-
}
192-
193183
Long createTime = channel2CreationDate.get(channel);
194184
if (createTime == null) {
195185
channel2CreationDate.putIfAbsent(channel, millisTime());
196186

197187
} else if (maxConnectionLifeTimeInMs != -1 && (createTime + maxConnectionLifeTimeInMs) < millisTime()) {
198-
log.debug("Channel {} expired", channel);
188+
LOGGER.debug("Channel {} expired", channel);
199189
return false;
200190
}
201191

202-
log.debug("Adding uri: {} for channel {}", uri, channel);
192+
LOGGER.debug("Adding uri: {} for channel {}", uri, channel);
203193
Channels.setDefaultAttribute(channel, DiscardEvent.INSTANCE);
204194

205-
ConcurrentLinkedQueue<IdleChannel> idleConnectionForHost = connectionsPool.get(uri);
206-
if (idleConnectionForHost == null) {
195+
ConcurrentLinkedQueue<IdleChannel> pooledConnectionForKey = connectionsPool.get(uri);
196+
if (pooledConnectionForKey == null) {
207197
ConcurrentLinkedQueue<IdleChannel> newPool = new ConcurrentLinkedQueue<IdleChannel>();
208-
idleConnectionForHost = connectionsPool.putIfAbsent(uri, newPool);
209-
if (idleConnectionForHost == null)
210-
idleConnectionForHost = newPool;
198+
pooledConnectionForKey = connectionsPool.putIfAbsent(uri, newPool);
199+
if (pooledConnectionForKey == null)
200+
pooledConnectionForKey = newPool;
211201
}
212202

213203
boolean added;
214-
int size = idleConnectionForHost.size();
204+
int size = pooledConnectionForKey.size();
215205
if (maxConnectionPerHost == -1 || size < maxConnectionPerHost) {
216206
IdleChannel idleChannel = new IdleChannel(uri, channel);
217-
synchronized (idleConnectionForHost) {
218-
added = idleConnectionForHost.add(idleChannel);
207+
synchronized (pooledConnectionForKey) {
208+
added = pooledConnectionForKey.add(idleChannel);
219209

220210
if (channel2IdleChannel.put(channel, idleChannel) != null) {
221-
log.error("Channel {} already exists in the connections pool!", channel);
211+
LOGGER.error("Channel {} already exists in the connections pool!", channel);
222212
}
223213
}
224214
} else {
225-
log.debug("Maximum number of requests per host reached {} for {}", maxConnectionPerHost, uri);
215+
LOGGER.debug("Maximum number of requests per host reached {} for {}", maxConnectionPerHost, uri);
226216
added = false;
227217
}
228218
return added;
@@ -237,13 +227,13 @@ public Channel poll(String uri) {
237227
}
238228

239229
IdleChannel idleChannel = null;
240-
ConcurrentLinkedQueue<IdleChannel> idleConnectionForHost = connectionsPool.get(uri);
241-
if (idleConnectionForHost != null) {
230+
ConcurrentLinkedQueue<IdleChannel> pooledConnectionForKey = connectionsPool.get(uri);
231+
if (pooledConnectionForKey != null) {
242232
boolean poolEmpty = false;
243233
while (!poolEmpty && idleChannel == null) {
244-
if (idleConnectionForHost.size() > 0) {
245-
synchronized (idleConnectionForHost) {
246-
idleChannel = idleConnectionForHost.poll();
234+
if (pooledConnectionForKey.size() > 0) {
235+
synchronized (pooledConnectionForKey) {
236+
idleChannel = pooledConnectionForKey.poll();
247237
if (idleChannel != null) {
248238
channel2IdleChannel.remove(idleChannel.channel);
249239
}
@@ -254,7 +244,7 @@ public Channel poll(String uri) {
254244
poolEmpty = true;
255245
} else if (!idleChannel.channel.isActive() || !idleChannel.channel.isOpen()) {
256246
idleChannel = null;
257-
log.trace("Channel not connected or not opened!");
247+
LOGGER.trace("Channel not connected or not opened!");
258248
}
259249
}
260250
}
@@ -266,12 +256,11 @@ private boolean remove(IdleChannel pooledChannel) {
266256
return false;
267257

268258
boolean isRemoved = false;
269-
ConcurrentLinkedQueue<IdleChannel> pooledConnectionForHost = connectionsPool.get(pooledChannel.uri);
259+
ConcurrentLinkedQueue<IdleChannel> pooledConnectionForHost = connectionsPool.get(pooledChannel.key);
270260
if (pooledConnectionForHost != null) {
271261
isRemoved = pooledConnectionForHost.remove(pooledChannel);
272262
}
273-
isRemoved |= channel2IdleChannel.remove(pooledChannel.channel) != null;
274-
return isRemoved;
263+
return isRemoved |= channel2IdleChannel.remove(pooledChannel.channel) != null;
275264
}
276265

277266
/**

0 commit comments

Comments
 (0)