Skip to content

Commit 6536825

Browse files
michaelklishinmichaelklishin
committed
merge default into bug25999
2 parents f19e850 + 235b399 commit 6536825

File tree

10 files changed

+151
-30
lines changed

10 files changed

+151
-30
lines changed

‎src/com/rabbitmq/client/ConnectionFactory.java‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232
import javax.net.ssl.TrustManager;
3333

3434
import com.rabbitmq.client.impl.AMQConnection;
35+
import com.rabbitmq.client.impl.DefaultThreadFactory;
3536
import com.rabbitmq.client.impl.ConnectionParams;
3637
import com.rabbitmq.client.impl.FrameHandler;
3738
import com.rabbitmq.client.impl.FrameHandlerFactory;
@@ -89,6 +90,7 @@ public class ConnectionFactory implements Cloneable {
8990
private SocketFactory factory = SocketFactory.getDefault();
9091
private SaslConfig saslConfig = DefaultSaslConfig.PLAIN;
9192
private ExecutorService sharedExecutor;
93+
private ThreadFactory threadFactory = new DefaultThreadFactory();
9294
private SocketConfigurator socketConf = new DefaultSocketConfigurator();
9395

9496
private boolean automaticRecovery = false;
@@ -426,6 +428,24 @@ public void setSharedExecutor(ExecutorService executor) {
426428
this.sharedExecutor = executor;
427429
}
428430

431+
/**
432+
* Retrieve the thread factory used to instantiate new threads.
433+
* @see com.rabbitmq.client.ThreadFactory
434+
* @see com.rabbitmq.client.impl.DefaultThreadFactory
435+
*/
436+
public ThreadFactory getThreadFactory() {
437+
return threadFactory;
438+
}
439+
440+
/**
441+
* Set the thread factory to use for newly created connections.
442+
* @see com.rabbitmq.client.ThreadFactory
443+
* @see com.rabbitmq.client.impl.DefaultThreadFactory
444+
*/
445+
public void setThreadFactory(ThreadFactory threadFactory) {
446+
this.threadFactory = threadFactory;
447+
}
448+
429449
public boolean isSSL(){
430450
return getSocketFactory() instanceof SSLSocketFactory;
431451
}
@@ -546,6 +566,7 @@ public Connection newConnection(ExecutorService executor, Address[] addrs)
546566
try {
547567
FrameHandler handler = fhFactory.create(addr);
548568
AMQConnection conn = new AMQConnection(params, handler);
569+
conn.setThreadFactory(this.threadFactory);
549570
conn.start();
550571
return conn;
551572
} catch (IOException e) {

‎src/com/rabbitmq/client/ThreadFactory.java‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
package com.rabbitmq.client;
2+
3+
/**
4+
* Extends {@link java.util.concurrent.ThreadFactory} to make it possible to specify
5+
* thread name.
6+
*
7+
* In environments with restricted thread management (e.g. Google App Engine), developers
8+
* can provide a custom factory to control how network I/O thread is created.
9+
*/
10+
public interface ThreadFactory extends java.util.concurrent.ThreadFactory {
11+
/**
12+
* Like {@link java.util.concurrent.ThreadFactory#newThread(Runnable)} but also takes
13+
* a thread name.
14+
*
15+
* @param r runnable to execute
16+
* @param threadName thread name
17+
* @return a new thread
18+
*/
19+
Thread newThread(Runnable r, String threadName);
20+
}

‎src/com/rabbitmq/client/impl/AMQConnection.java‎

Lines changed: 48 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import java.util.HashMap;
2727
import java.util.Map;
2828
import java.util.concurrent.CopyOnWriteArrayList;
29+
import java.util.concurrent.ExecutorService;
2930
import java.util.concurrent.TimeoutException;
3031

3132
import com.rabbitmq.client.AMQP;
@@ -44,6 +45,7 @@
4445
import com.rabbitmq.client.SaslConfig;
4546
import com.rabbitmq.client.SaslMechanism;
4647
import com.rabbitmq.client.ShutdownSignalException;
48+
import com.rabbitmq.client.ThreadFactory;
4749
import com.rabbitmq.client.impl.AMQChannel.BlockingRpcContinuation;
4850
import com.rabbitmq.utility.BlockingCell;
4951

@@ -61,6 +63,9 @@ final class Copyright {
6163
public class AMQConnection extends ShutdownNotifierComponent implements Connection, NetworkConnection {
6264
/** Timeout used while waiting for AMQP handshaking to complete (milliseconds) */
6365
public static final int HANDSHAKE_TIMEOUT = 10000;
66+
private final ExecutorService executor;
67+
private Thread mainLoopThread;
68+
private ThreadFactory threadFactory = new DefaultThreadFactory();
6469

6570
/**
6671
* Retrieve a copy of the default table of client properties that
@@ -101,7 +106,7 @@ public static final Map<String, Object> defaultClientProperties() {
101106
}
102107
};
103108

104-
protected final ConsumerWorkService _workService;
109+
protected ConsumerWorkService _workService = null;
105110

106111
/** Frame source/sink */
107112
private final FrameHandler _frameHandler;
@@ -124,7 +129,7 @@ public static final Map<String, Object> defaultClientProperties() {
124129
private volatile boolean _inConnectionNegotiation;
125130

126131
/** Manages heart-beat sending for this connection */
127-
private final HeartbeatSender _heartbeatSender;
132+
private HeartbeatSender _heartbeatSender;
128133

129134
private final String _virtualHost;
130135
private final Map<String, Object> _clientProperties;
@@ -212,16 +217,23 @@ public AMQConnection(ConnectionParams params, FrameHandler frameHandler)
212217
this.requestedChannelMax = params.getRequestedChannelMax();
213218
this.requestedHeartbeat = params.getRequestedHeartbeat();
214219
this.saslConfig = params.getSaslConfig();
220+
this.executor = params.getExecutor();
215221

216-
this._workService = new ConsumerWorkService(params.getExecutor());
217222
this._channelManager = null;
218223

219-
this._heartbeatSender = new HeartbeatSender(frameHandler);
220224
this._brokerInitiatedShutdown = false;
221225

222226
this._inConnectionNegotiation = true; // we start out waiting for the first protocol response
223227
}
224228

229+
private void initializeConsumerWorkService() {
230+
this._workService = new ConsumerWorkService(executor, threadFactory);
231+
}
232+
233+
private void initializeHeartbeatSender() {
234+
this._heartbeatSender = new HeartbeatSender(_frameHandler, threadFactory);
235+
}
236+
225237
/**
226238
* Start up the connection, including the MainLoop thread.
227239
* Sends the protocol
@@ -241,6 +253,8 @@ public AMQConnection(ConnectionParams params, FrameHandler frameHandler)
241253
public void start()
242254
throws IOException
243255
{
256+
initializeConsumerWorkService();
257+
initializeHeartbeatSender();
244258
this._running = true;
245259
// Make sure that the first thing we do is to send the header,
246260
// which should cause any socket errors to show up for us, rather
@@ -264,7 +278,9 @@ public void start()
264278
}
265279

266280
// start the main loop going
267-
new MainLoop("AMQP Connection " + getHostAddress() + ":" + getPort()).start();
281+
MainLoop loop = new MainLoop();
282+
mainLoopThread = threadFactory.newThread(loop, "AMQP Connection " + getHostAddress() + ":" + getPort());
283+
mainLoopThread.start();
268284
// after this point clear-up of MainLoop is triggered by closing the frameHandler.
269285

270286
AMQP.Connection.Start connStart = null;
@@ -337,7 +353,7 @@ public void start()
337353
int channelMax =
338354
negotiateChannelMax(this.requestedChannelMax,
339355
connTune.getChannelMax());
340-
_channelManager = instantiateChannelManager(channelMax);
356+
_channelManager = instantiateChannelManager(channelMax, threadFactory);
341357

342358
int frameMax =
343359
negotiatedMaxValue(this.requestedFrameMax,
@@ -374,8 +390,8 @@ public void start()
374390
return;
375391
}
376392

377-
protected ChannelManager instantiateChannelManager(int channelMax) {
378-
return new ChannelManager(this._workService, channelMax);
393+
protected ChannelManager instantiateChannelManager(int channelMax, ThreadFactory threadFactory) {
394+
return new ChannelManager(this._workService, channelMax, threadFactory);
379395
}
380396

381397
/**
@@ -427,6 +443,23 @@ public void setHeartbeat(int heartbeat) {
427443
}
428444
}
429445

446+
/**
447+
* Makes it possible to override thread factory that is used
448+
* to instantiate connection network I/O loop. Only necessary
449+
* in the environments with restricted
450+
* @param threadFactory
451+
*/
452+
public void setThreadFactory(ThreadFactory threadFactory) {
453+
this.threadFactory = threadFactory;
454+
}
455+
456+
/**
457+
* @return Thread factory used by this connection.
458+
*/
459+
public ThreadFactory getThreadFactory() {
460+
return threadFactory;
461+
}
462+
430463
public Map<String, Object> getClientProperties() {
431464
return new HashMap<String, Object>(_clientProperties);
432465
}
@@ -485,14 +518,7 @@ private static final int negotiatedMaxValue(int clientValue, int serverValue) {
485518
Math.min(clientValue, serverValue);
486519
}
487520

488-
private class MainLoop extends Thread {
489-
490-
/**
491-
* @param name of thread
492-
*/
493-
MainLoop(String name) {
494-
super(name);
495-
}
521+
private class MainLoop implements Runnable {
496522

497523
/**
498524
* Channel reader thread main loop. Reads a frame, and if it is
@@ -636,13 +662,14 @@ public void handleConnectionClose(Command closeCommand) {
636662
_channel0.quiescingTransmit(new AMQP.Connection.CloseOk.Builder().build());
637663
} catch (IOException _) { } // ignore
638664
_brokerInitiatedShutdown = true;
639-
Thread scw = new SocketCloseWait(sse);
640-
scw.setName("AMQP Connection Closing Monitor " +
665+
SocketCloseWait scw = new SocketCloseWait(sse);
666+
Thread waiter = threadFactory.newThread(scw);
667+
waiter.setName("AMQP Connection Closing Monitor " +
641668
getHostAddress() + ":" + getPort());
642-
scw.start();
669+
waiter.start();
643670
}
644671

645-
private class SocketCloseWait extends Thread {
672+
private class SocketCloseWait implements Runnable {
646673
private final ShutdownSignalException cause;
647674

648675
public SocketCloseWait(ShutdownSignalException sse) {
@@ -789,7 +816,7 @@ public void close(int closeCode,
789816
boolean abort)
790817
throws IOException
791818
{
792-
boolean sync = !(Thread.currentThread() instanceof MainLoop);
819+
boolean sync = !(Thread.currentThread() == mainLoopThread);
793820

794821
try {
795822
AMQP.Connection.Close reason =

‎src/com/rabbitmq/client/impl/ChannelManager.java‎

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import java.util.concurrent.TimeUnit;
2626

2727
import com.rabbitmq.client.ShutdownSignalException;
28+
import com.rabbitmq.client.ThreadFactory;
2829
import com.rabbitmq.utility.IntAllocator;
2930

3031
/**
@@ -45,12 +46,17 @@ public class ChannelManager {
4546

4647
/** Maximum channel number available on this connection. */
4748
private final int _channelMax;
49+
private final ThreadFactory threadFactory;
4850

4951
public int getChannelMax(){
5052
return _channelMax;
5153
}
5254

5355
public ChannelManager(ConsumerWorkService workService, int channelMax) {
56+
this(workService, channelMax, new DefaultThreadFactory());
57+
}
58+
59+
public ChannelManager(ConsumerWorkService workService, int channelMax, ThreadFactory threadFactory) {
5460
if (channelMax == 0) {
5561
// The framing encoding only allows for unsigned 16-bit integers
5662
// for the channel number
@@ -60,6 +66,7 @@ public ChannelManager(ConsumerWorkService workService, int channelMax) {
6066
channelNumberAllocator = new IntAllocator(1, channelMax);
6167

6268
this.workService = workService;
69+
this.threadFactory = threadFactory;
6370
}
6471

6572
/**
@@ -97,14 +104,18 @@ public void handleSignal(ShutdownSignalException signal) {
97104
private void scheduleShutdownProcessing() {
98105
final Set<CountDownLatch> sdSet = new HashSet<CountDownLatch>(shutdownSet);
99106
final ConsumerWorkService ssWorkService = workService;
100-
Thread shutdownThread = new Thread( new Runnable() {
107+
Runnable target = new Runnable() {
101108
public void run() {
102109
for (CountDownLatch latch : sdSet) {
103110
try { latch.await(SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS); } catch (Throwable e) { /*ignored*/ }
104111
}
105112
ssWorkService.shutdown();
106-
}}, "ConsumerWorkServiceShutdown");
107-
shutdownThread.setDaemon(true);
113+
}
114+
};
115+
Thread shutdownThread = threadFactory.newThread(target, "ConsumerWorkService shutdown monitor");
116+
if(Environment.isAllowedToModifyThreads()) {
117+
shutdownThread.setDaemon(true);
118+
}
108119
shutdownThread.start();
109120
}
110121

‎src/com/rabbitmq/client/impl/ConsumerWorkService.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import java.util.List;
2020
import java.util.concurrent.ExecutorService;
2121
import java.util.concurrent.Executors;
22+
import java.util.concurrent.ThreadFactory;
2223

2324
import com.rabbitmq.client.Channel;
2425

@@ -29,9 +30,9 @@ final public class ConsumerWorkService {
2930
private final boolean privateExecutor;
3031
private final WorkPool<Channel, Runnable> workPool;
3132

32-
public ConsumerWorkService(ExecutorService executor) {
33+
public ConsumerWorkService(ExecutorService executor, ThreadFactory threadFactory) {
3334
this.privateExecutor = (executor == null);
34-
this.executor = (executor == null) ? Executors.newFixedThreadPool(DEFAULT_NUM_THREADS)
35+
this.executor = (executor == null) ? Executors.newFixedThreadPool(DEFAULT_NUM_THREADS, threadFactory)
3536
: executor;
3637
this.workPool = new WorkPool<Channel, Runnable>();
3738
}

‎src/com/rabbitmq/client/impl/DefaultThreadFactory.java‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
package com.rabbitmq.client.impl;
2+
3+
import com.rabbitmq.client.ThreadFactory;
4+
5+
/**
6+
* Default thread factory that instantiates {@link java.lang.Thread}s directly.
7+
*/
8+
public class DefaultThreadFactory implements ThreadFactory {
9+
@Override
10+
public Thread newThread(Runnable r, String threadName) {
11+
return new Thread(r, threadName);
12+
}
13+
14+
@Override
15+
public Thread newThread(Runnable r) {
16+
return new Thread(r);
17+
}
18+
}

‎src/com/rabbitmq/client/impl/Environment.java‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
package com.rabbitmq.client.impl;
2+
3+
/**
4+
* Infers information about the execution environment, e.g.
5+
* security permissions.
6+
*/
7+
class Environment {
8+
public static boolean isAllowedToModifyThreads() {
9+
try {
10+
SecurityManager sm = new SecurityManager();
11+
sm.checkPermission(new RuntimePermission("modifyThread"));
12+
sm.checkPermission(new RuntimePermission("modifyThreadGroup"));
13+
return true;
14+
} catch (SecurityException se) {
15+
return false;
16+
}
17+
}
18+
}

‎src/com/rabbitmq/client/impl/HeartbeatSender.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import java.util.concurrent.ExecutorService;
2323
import java.util.concurrent.ScheduledExecutorService;
2424
import java.util.concurrent.Executors;
25+
import java.util.concurrent.ThreadFactory;
2526
import java.util.concurrent.TimeUnit;
2627
import java.util.concurrent.ScheduledFuture;
2728
import java.io.IOException;
@@ -39,6 +40,7 @@ final class HeartbeatSender {
3940
private final Object monitor = new Object();
4041

4142
private final FrameHandler frameHandler;
43+
private final ThreadFactory threadFactory;
4244

4345
private ScheduledExecutorService executor;
4446

@@ -48,8 +50,9 @@ final class HeartbeatSender {
4850

4951
private volatile long lastActivityTime;
5052

51-
HeartbeatSender(FrameHandler frameHandler) {
53+
HeartbeatSender(FrameHandler frameHandler, ThreadFactory threadFactory) {
5254
this.frameHandler = frameHandler;
55+
this.threadFactory = threadFactory;
5356
}
5457

5558
public void signalActivity() {
@@ -86,7 +89,7 @@ public void setHeartbeat(int heartbeatSeconds) {
8689
private ScheduledExecutorService createExecutorIfNecessary() {
8790
synchronized (this.monitor) {
8891
if (this.executor == null) {
89-
this.executor = Executors.newSingleThreadScheduledExecutor();
92+
this.executor = Executors.newSingleThreadScheduledExecutor(threadFactory);
9093
}
9194
return this.executor;
9295
}

0 commit comments

Comments
 (0)