{
final Configuration config = server.getConfiguration();
Executor connectionExecutor = server.getExecutorFactory().getExecutor();
final CoreRemotingConnection rc = new RemotingConnectionImpl(connection,
interceptors,
config.isAsyncConnectionExecutionEnabled() ? connectionExecutor
: null,
server.getNodeID());
Channel channel1 = rc.getChannel(1, -1);
ChannelHandler handler = new HornetQPacketHandler(this, server, channel1, rc);
channel1.setHandler(handler);
long ttl = HornetQClient.DEFAULT_CONNECTION_TTL;
if (config.getConnectionTTLOverride() != -1)
{
ttl = config.getConnectionTTLOverride();
}
final ConnectionEntry entry = new ConnectionEntry(rc, connectionExecutor, System.currentTimeMillis(), ttl);
final Channel channel0 = rc.getChannel(0, -1);
channel0.setHandler(new ChannelHandler()
{
public void handlePacket(final Packet packet)
{
if (packet.getType() == PacketImpl.PING)
{
Ping ping = (Ping)packet;
if (config.getConnectionTTLOverride() == -1)
{
// Allow clients to specify connection ttl
entry.ttl = ping.getConnectionTTL();
}
// Just send a ping back
channel0.send(packet);
}
else if (packet.getType() == PacketImpl.SUBSCRIBE_TOPOLOGY || packet.getType() == PacketImpl.SUBSCRIBE_TOPOLOGY_V2)
{
SubscribeClusterTopologyUpdatesMessage msg = (SubscribeClusterTopologyUpdatesMessage)packet;
if (packet.getType() == PacketImpl.SUBSCRIBE_TOPOLOGY_V2)
{
channel0.getConnection().setClientVersion(((SubscribeClusterTopologyUpdatesMessageV2)msg).getClientVersion());
}
final ClusterTopologyListener listener = new ClusterTopologyListener()
{
public void nodeUP(final long uniqueEventID,
final String nodeID,
final Pair<TransportConfiguration, TransportConfiguration> connectorPair,
final boolean last)
{
// Using an executor as most of the notifications on the Topology
// may come from a channel itself
// What could cause deadlocks
entry.connectionExecutor.execute(new Runnable()
{
public void run()
{
if (channel0.supports(PacketImpl.CLUSTER_TOPOLOGY_V2))
{
channel0.send(new ClusterTopologyChangeMessage_V2(uniqueEventID, nodeID, connectorPair, last));
}
else
{
channel0.send(new ClusterTopologyChangeMessage(nodeID, connectorPair, last));
}
}
});
}
public void nodeDown(final long uniqueEventID, final String nodeID)
{
// Using an executor as most of the notifications on the Topology
// may come from a channel itself
// What could cause deadlocks
entry.connectionExecutor.execute(new Runnable()
{
public void run()
{
if (channel0.supports(PacketImpl.CLUSTER_TOPOLOGY_V2))
{
channel0.send(new ClusterTopologyChangeMessage_V2(uniqueEventID, nodeID));
}
else
{
channel0.send(new ClusterTopologyChangeMessage(nodeID));
}
}
});
}
public String toString()
{
return "Remote Proxy on channel " + Integer.toHexString(System.identityHashCode(this));
}
};
final boolean isCC = msg.isClusterConnection();
if (acceptorUsed.getClusterConnection() != null)
{
acceptorUsed.getClusterConnection().addClusterTopologyListener(listener, isCC);
rc.addCloseListener(new CloseListener()
{
public void connectionClosed()
{
acceptorUsed.getClusterConnection().removeClusterTopologyListener(listener, isCC);
}