[hornetq-commits] JBoss hornetq SVN: r10905 - trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl.
do-not-reply at jboss.org
do-not-reply at jboss.org
Fri Jul 1 08:23:52 EDT 2011
Author: borges
Date: 2011-07-01 08:23:52 -0400 (Fri, 01 Jul 2011)
New Revision: 10905
Modified:
trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/ChannelImpl.java
trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/RemotingConnectionImpl.java
Log:
Place standard channel numbers into constants (better than at code comment)
Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/ChannelImpl.java
===================================================================
--- trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/ChannelImpl.java 2011-07-01 12:23:16 UTC (rev 10904)
+++ trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/ChannelImpl.java 2011-07-01 12:23:52 UTC (rev 10905)
@@ -37,9 +37,33 @@
*/
public class ChannelImpl implements Channel
{
+ public enum CHANNEL_ID {
+
+ /**
+ * Used for core protocol management.
+ * @see CoreProtocolManager
+ */
+ PING(0),
+ /** Session creation and attachment. */
+ SESSION(1),
+ /** Replication, i.e. for backups that do not share the journal. */
+ REPLICATION(2),
+ /**
+ * Channels [0-9] are reserved for the system, user channels must be greater than that.
+ */
+ USER(10);
+
+ public final long id;
+
+ CHANNEL_ID(long id)
+ {
+ this.id = id;
+ }
+ }
+
private static final Logger log = Logger.getLogger(ChannelImpl.class);
- private volatile long id;
+ private final long id;
private ChannelHandler handler;
@@ -336,7 +360,7 @@
// And switch it
- final CoreRemotingConnection rnewConnection = (CoreRemotingConnection)newConnection;
+ final CoreRemotingConnection rnewConnection = newConnection;
rnewConnection.putChannel(id, this);
Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/RemotingConnectionImpl.java
===================================================================
--- trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/RemotingConnectionImpl.java 2011-07-01 12:23:16 UTC (rev 10904)
+++ trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/RemotingConnectionImpl.java 2011-07-01 12:23:52 UTC (rev 10905)
@@ -30,6 +30,7 @@
import org.hornetq.core.protocol.core.Channel;
import org.hornetq.core.protocol.core.CoreRemotingConnection;
import org.hornetq.core.protocol.core.Packet;
+import org.hornetq.core.protocol.core.impl.ChannelImpl.CHANNEL_ID;
import org.hornetq.core.protocol.core.impl.wireformat.DisconnectMessage;
import org.hornetq.core.remoting.CloseListener;
import org.hornetq.core.remoting.FailureListener;
@@ -71,11 +72,7 @@
private final boolean client;
- // Channels 0-9 are reserved for the system
- // 0 is for pinging
- // 1 is for session creation and attachment
- // 2 is for replication
- private volatile SimpleIDGenerator idGenerator = new SimpleIDGenerator(10);
+ private volatile SimpleIDGenerator idGenerator = new SimpleIDGenerator(CHANNEL_ID.USER.id);
private boolean idGeneratorSynced = false;
@@ -88,13 +85,13 @@
private volatile boolean dataReceived;
private final Executor executor;
-
+
private volatile boolean executing;
-
+
private final SimpleString nodeID;
private final long creationTime;
-
+
private String clientID;
// Constructors
@@ -139,9 +136,9 @@
this.client = client;
this.executor = executor;
-
+
this.nodeID = nodeID;
-
+
this.creationTime = System.currentTimeMillis();
}
@@ -174,7 +171,7 @@
{
return transportConnection.getRemoteAddress();
}
-
+
public long getCreationTime()
{
return creationTime;
@@ -247,26 +244,26 @@
public List<CloseListener> removeCloseListeners()
{
List<CloseListener> ret = new ArrayList<CloseListener>(closeListeners);
-
+
closeListeners.clear();
-
+
return ret;
}
public List<FailureListener> removeFailureListeners()
{
List<FailureListener> ret = new ArrayList<FailureListener>(failureListeners);
-
+
failureListeners.clear();
-
- return ret;
+
+ return ret;
}
public void setCloseListeners(List<CloseListener> listeners)
{
closeListeners.clear();
-
- closeListeners.addAll(listeners);
+
+ closeListeners.addAll(listeners);
}
public HornetQBuffer createBuffer(final int size)
@@ -323,7 +320,7 @@
callClosingListeners();
}
-
+
public void disconnect()
{
Channel channel0 = getChannel(0, -1);
@@ -331,13 +328,13 @@
// And we remove all channels from the connection, this ensures no more packets will be processed after this
// method is
// complete
-
+
Set<Channel> allChannels = new HashSet<Channel>(channels.values());
removeAllChannels();
// Now we are 100% sure that no more packets will be processed we can flush then send the disconnect
-
+
for (Channel channel: allChannels)
{
channel.flushConfirmations();
@@ -408,7 +405,7 @@
}
}
}
-
+
public void checkFlushBatchBuffer()
{
transportConnection.checkFlushBatchBuffer();
@@ -422,11 +419,11 @@
try
{
final Packet packet = decoder.decode(buffer);
-
+
if (packet.isAsyncExec() && executor != null)
{
executing = true;
-
+
executor.execute(new Runnable()
{
public void run()
@@ -439,7 +436,7 @@
{
RemotingConnectionImpl.log.error("Unexpected error", t);
}
-
+
executing = false;
}
});
@@ -451,13 +448,13 @@
{
Thread.yield();
}
-
+
// Pings must always be handled out of band so we can send pings back to the client quickly
// otherwise they would get in the queue with everything else which might give an intolerable delay
doBufferReceived(packet);
}
-
- dataReceived = true;
+
+ dataReceived = true;
}
catch (Exception e)
{
@@ -515,8 +512,8 @@
{
channels.clear();
}
- }
-
+ }
+
private void callFailureListeners(final HornetQException me)
{
final List<FailureListener> listenersClone = new ArrayList<FailureListener>(failureListeners);
@@ -572,7 +569,7 @@
{
clientID = cID;
}
-
+
public String getClientID()
{
return clientID;
More information about the hornetq-commits
mailing list