mirror of
https://github.com/OpenHFT/Chronicle-Map.git
synced 2021-09-13 01:26:12 +00:00
HCOLL-134 Broadcast active ChronicleMap host names over UDP to allow other maps to connect
This commit is contained in:
Rob Austin
@@ -31,8 +31,6 @@ import java.nio.channels.ClosedChannelException;
|
||||
import java.nio.channels.SelectableChannel;
|
||||
import java.nio.channels.SelectionKey;
|
||||
import java.nio.channels.Selector;
|
||||
import java.util.Collections;
|
||||
import java.util.LinkedHashSet;
|
||||
import java.util.Queue;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.ConcurrentLinkedQueue;
|
||||
@@ -186,7 +184,6 @@ abstract class AbstractChannelReplicator implements Closeable {
|
||||
lastTime = time;
|
||||
bytesWritten = 0;
|
||||
|
||||
|
||||
if (LOG.isDebugEnabled())
|
||||
LOG.debug("Restoring OP_WRITE on all channels");
|
||||
|
||||
|
||||
@@ -28,7 +28,6 @@ import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.io.*;
|
||||
import java.nio.channels.FileChannel;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
@@ -0,0 +1,291 @@
|
||||
/*
|
||||
* Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package net.openhft.chronicle.map;
|
||||
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.io.IOException;
|
||||
import java.net.InetAddress;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.NetworkInterface;
|
||||
import java.net.StandardSocketOptions;
|
||||
import java.nio.channels.*;
|
||||
import java.util.Set;
|
||||
|
||||
import static java.net.StandardProtocolFamily.INET;
|
||||
import static java.net.StandardProtocolFamily.INET6;
|
||||
import static java.net.StandardSocketOptions.IP_MULTICAST_IF;
|
||||
import static java.net.StandardSocketOptions.SO_REUSEADDR;
|
||||
import static java.nio.channels.SelectionKey.OP_READ;
|
||||
import static java.nio.channels.SelectionKey.OP_WRITE;
|
||||
|
||||
|
||||
/**
|
||||
* The UdpReplicator attempts to read the data ( but it does not enforce or grantee delivery ), typically, you
|
||||
* should use the UdpReplicator if you have a large number of nodes, and you wish to receive the data before
|
||||
* it becomes available on TCP/IP. In order to not miss data, UdpReplicator should be used in conjunction with
|
||||
* the TCP Replicator.
|
||||
*/
|
||||
class UdpChannelReplicator extends AbstractChannelReplicator implements Replica.ModificationNotifier, Closeable {
|
||||
|
||||
private static final Logger LOG =
|
||||
LoggerFactory.getLogger(UdpChannelReplicator.class.getName());
|
||||
|
||||
private final byte localIdentifier;
|
||||
private EntryWriter writer;
|
||||
private EntryReader reader;
|
||||
|
||||
private final InetAddress address;
|
||||
private final int port;
|
||||
private final NetworkInterface networkInterface;
|
||||
private final ServerConnector serverConnector;
|
||||
|
||||
private SelectableChannel writeChannel;
|
||||
private volatile boolean shouldEnableOpWrite;
|
||||
|
||||
/**
|
||||
* @param replicationConfig
|
||||
* @param maxPayloadSize the maximum size of any serialized entry
|
||||
* @param localIdentifier just used for logging
|
||||
* @throws IOException
|
||||
*/
|
||||
UdpChannelReplicator(@NotNull final UdpReplicationConfig replicationConfig,
|
||||
final int maxPayloadSize, final byte localIdentifier)
|
||||
throws IOException {
|
||||
|
||||
super("UdpReplicator-" + localIdentifier, replicationConfig.throttlingConfig(),
|
||||
maxPayloadSize);
|
||||
|
||||
this.localIdentifier = localIdentifier;
|
||||
|
||||
address = replicationConfig.address();
|
||||
port = replicationConfig.port();
|
||||
networkInterface = replicationConfig.networkInterface();
|
||||
serverConnector = new ServerConnector();
|
||||
}
|
||||
|
||||
|
||||
void setWriter(EntryWriter writer) {
|
||||
this.writer = writer;
|
||||
}
|
||||
|
||||
void setReader(EntryReader reader) {
|
||||
this.reader = reader;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
writeChannel = null;
|
||||
super.close();
|
||||
}
|
||||
|
||||
/**
|
||||
* binds to the server socket and process data This method will block until interrupted
|
||||
*/
|
||||
@Override
|
||||
void process() throws IOException {
|
||||
|
||||
connectClient().register(selector, OP_READ);
|
||||
serverConnector.connectLater();
|
||||
|
||||
while (selector.isOpen()) {
|
||||
|
||||
registerPendingRegistrations();
|
||||
|
||||
// this may block for a long time, upon return the
|
||||
// selected set contains keys of the ready channels
|
||||
final int n = selector.select(100);
|
||||
|
||||
if (shouldEnableOpWrite)
|
||||
enableWrites();
|
||||
|
||||
checkThrottleInterval();
|
||||
|
||||
if (n == 0) {
|
||||
continue; // nothing to do
|
||||
}
|
||||
|
||||
final Set<SelectionKey> selectionKeys = selector.selectedKeys();
|
||||
for (final SelectionKey key : selectionKeys) {
|
||||
|
||||
try {
|
||||
|
||||
if (key.isReadable()) {
|
||||
final DatagramChannel socketChannel = (DatagramChannel) key.channel();
|
||||
reader.readAll(socketChannel);
|
||||
}
|
||||
|
||||
if (key.isWritable()) {
|
||||
final DatagramChannel socketChannel = (DatagramChannel) key.channel();
|
||||
try {
|
||||
int bytesJustWritten = writer.writeAll(socketChannel);
|
||||
contemplateThrottleWrites(bytesJustWritten);
|
||||
} catch (NotYetConnectedException e) {
|
||||
if (LOG.isDebugEnabled())
|
||||
LOG.debug("", e);
|
||||
serverConnector.connectLater();
|
||||
} catch (IOException e) {
|
||||
if (LOG.isDebugEnabled())
|
||||
LOG.debug("", e);
|
||||
serverConnector.connectLater();
|
||||
}
|
||||
}
|
||||
|
||||
} catch (Exception e) {
|
||||
LOG.error("", e);
|
||||
closeEarlyAndQuietly(key.channel());
|
||||
}
|
||||
}
|
||||
|
||||
selectionKeys.clear();
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
|
||||
private DatagramChannel connectClient() throws IOException {
|
||||
final DatagramChannel client = address.isMulticastAddress() ?
|
||||
DatagramChannel.open(address.getAddress().length == 4 ? INET : INET6) :
|
||||
DatagramChannel.open();
|
||||
|
||||
final InetSocketAddress hostAddress = new InetSocketAddress(port);
|
||||
client.configureBlocking(false);
|
||||
if (address.isMulticastAddress()) {
|
||||
client.setOption(IP_MULTICAST_IF, networkInterface);
|
||||
client.setOption(SO_REUSEADDR, true);
|
||||
client.bind(hostAddress);
|
||||
client.join(address, networkInterface);
|
||||
if (LOG.isDebugEnabled())
|
||||
LOG.debug("Connecting via multicast, group=" + address);
|
||||
} else {
|
||||
client.bind(hostAddress);
|
||||
}
|
||||
|
||||
if (LOG.isDebugEnabled())
|
||||
LOG.debug("Listening on port " + port);
|
||||
closeables.add(client);
|
||||
return client;
|
||||
}
|
||||
|
||||
/**
|
||||
* called whenever there is a change to the modification iterator
|
||||
*/
|
||||
@Override
|
||||
public void onChange() {
|
||||
// the write have to be enabled on the same thread as the selector
|
||||
shouldEnableOpWrite = true;
|
||||
selector.wakeup();
|
||||
}
|
||||
|
||||
private void enableWrites() {
|
||||
|
||||
if (writeChannel == null)
|
||||
return;
|
||||
|
||||
try {
|
||||
final SelectionKey selectionKey = writeChannel.keyFor(this.selector);
|
||||
if (selectionKey != null)
|
||||
selectionKey.interestOps(selectionKey.interestOps() | OP_WRITE);
|
||||
} catch (Exception e) {
|
||||
LOG.error("", e);
|
||||
}
|
||||
}
|
||||
|
||||
void disableWrites() {
|
||||
|
||||
if (writeChannel == null)
|
||||
return;
|
||||
|
||||
try {
|
||||
final SelectionKey selectionKey = writeChannel.keyFor(this.selector);
|
||||
selectionKey.interestOps(selectionKey.interestOps() & ~OP_WRITE);
|
||||
} catch (Exception e) {
|
||||
LOG.error("", e);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private class ServerConnector extends AbstractConnector {
|
||||
private final InetSocketAddress socketAddress;
|
||||
|
||||
private ServerConnector() {
|
||||
super("UDP-Connector");
|
||||
this.socketAddress = new InetSocketAddress(address, port);
|
||||
}
|
||||
|
||||
SelectableChannel doConnect() throws
|
||||
IOException, InterruptedException {
|
||||
|
||||
final DatagramChannel server = DatagramChannel.open();
|
||||
server.configureBlocking(false);
|
||||
|
||||
// Kick off connection establishment
|
||||
try {
|
||||
// Create a non-blocking socket channel
|
||||
server.socket().setBroadcast(true);
|
||||
server.connect(socketAddress);
|
||||
} catch (IOException e) {
|
||||
if (LOG.isDebugEnabled())
|
||||
LOG.debug("details=" + new Details(socketAddress, localIdentifier), e);
|
||||
connectLater();
|
||||
return null;
|
||||
}
|
||||
|
||||
server.setOption(SO_REUSEADDR, true)
|
||||
.setOption(StandardSocketOptions.IP_MULTICAST_LOOP, false)
|
||||
.setOption(StandardSocketOptions.SO_BROADCAST, true)
|
||||
.setOption(SO_REUSEADDR, true);
|
||||
|
||||
// the registration has be be run on the same thread as the selector
|
||||
addPendingRegistration(new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
|
||||
try {
|
||||
server.register(selector, OP_WRITE);
|
||||
writeChannel = server;
|
||||
throttle(server);
|
||||
} catch (ClosedChannelException e) {
|
||||
LOG.error("", e);
|
||||
}
|
||||
|
||||
}
|
||||
});
|
||||
|
||||
return server;
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
interface EntryReader {
|
||||
void readAll(@NotNull final DatagramChannel socketChannel) throws IOException, InterruptedException;
|
||||
}
|
||||
|
||||
interface EntryWriter {
|
||||
int writeAll(@NotNull final DatagramChannel socketChannel) throws InterruptedException, IOException;
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -23,203 +23,43 @@ import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.io.IOException;
|
||||
import java.net.InetAddress;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.NetworkInterface;
|
||||
import java.net.StandardSocketOptions;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.*;
|
||||
import java.util.Set;
|
||||
|
||||
import static java.net.StandardProtocolFamily.INET;
|
||||
import static java.net.StandardProtocolFamily.INET6;
|
||||
import static java.net.StandardSocketOptions.IP_MULTICAST_IF;
|
||||
import static java.net.StandardSocketOptions.SO_REUSEADDR;
|
||||
import static java.nio.channels.SelectionKey.OP_READ;
|
||||
import static java.nio.channels.SelectionKey.OP_WRITE;
|
||||
import java.nio.channels.DatagramChannel;
|
||||
|
||||
|
||||
/**
|
||||
* The UdpReplicator attempts to read the data ( but it does not enforce or grantee delivery ),
|
||||
* typically, you should use the UdpReplicator if you have a large number of nodes,
|
||||
* and you wish to receive the data before it becomes available on TCP/IP.
|
||||
* In order to not miss data, UdpReplicator should be used in conjunction with the TCP Replicator.
|
||||
* The UdpReplicator attempts to read the data ( but it does not enforce or grantee delivery ), typically, you
|
||||
* should use the UdpReplicator if you have a large number of nodes, and you wish to receive the data before
|
||||
* it becomes available on TCP/IP. In order to not miss data, UdpReplicator should be used in conjunction with
|
||||
* the TCP Replicator.
|
||||
*/
|
||||
class UdpReplicator extends AbstractChannelReplicator implements Replica.ModificationNotifier, Closeable {
|
||||
class UdpReplicator extends UdpChannelReplicator implements Replica.ModificationNotifier, Closeable {
|
||||
|
||||
private static final Logger LOG =
|
||||
LoggerFactory.getLogger(UdpReplicator.class.getName());
|
||||
|
||||
private final byte localIdentifier;
|
||||
private final UdpSocketChannelEntryWriter writer;
|
||||
private final UdpSocketChannelEntryReader reader;
|
||||
private final Replica replica;
|
||||
private final InetAddress address;
|
||||
private final int port;
|
||||
private final NetworkInterface networkInterface;
|
||||
private final ServerConnector serverConnector;
|
||||
private Replica.ModificationIterator modificationIterator;
|
||||
private SelectableChannel writeChannel;
|
||||
private volatile boolean shouldEnableOpWrite;
|
||||
|
||||
UdpReplicator(@NotNull final Replica replica,
|
||||
@NotNull final Replica.EntryExternalizable entryExternalizable,
|
||||
@NotNull final UdpReplicationConfig replicationConfig,
|
||||
final int serializedEntrySize)
|
||||
throws IOException {
|
||||
super("UdpReplicator-" + replica.identifier(), replicationConfig.throttlingConfig(),
|
||||
serializedEntrySize);
|
||||
|
||||
this.localIdentifier = replica.identifier();
|
||||
this.replica = replica;
|
||||
this.writer = new UdpSocketChannelEntryWriter(serializedEntrySize, entryExternalizable);
|
||||
this.reader = new UdpSocketChannelEntryReader(serializedEntrySize, entryExternalizable);
|
||||
super(replicationConfig, serializedEntrySize, replica.identifier());
|
||||
|
||||
address = replicationConfig.address();
|
||||
port = replicationConfig.port();
|
||||
networkInterface = replicationConfig.networkInterface();
|
||||
|
||||
serverConnector = new ServerConnector();
|
||||
|
||||
modificationIterator = replica.acquireModificationIterator(
|
||||
Replica.ModificationIterator modificationIterator = replica.acquireModificationIterator(
|
||||
ChronicleMapBuilder.UDP_REPLICATION_MODIFICATION_ITERATOR_ID, this);
|
||||
|
||||
setReader(new UdpSocketChannelEntryReader(serializedEntrySize, entryExternalizable));
|
||||
|
||||
setWriter(new UdpSocketChannelEntryWriter(serializedEntrySize, entryExternalizable,
|
||||
modificationIterator, this));
|
||||
|
||||
start();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
writeChannel = null;
|
||||
super.close();
|
||||
}
|
||||
|
||||
/**
|
||||
* binds to the server socket and process data This method will block until interrupted
|
||||
*/
|
||||
@Override
|
||||
void process() throws IOException {
|
||||
|
||||
connectClient().register(selector, OP_READ);
|
||||
serverConnector.connectLater();
|
||||
|
||||
while (selector.isOpen()) {
|
||||
|
||||
registerPendingRegistrations();
|
||||
|
||||
// this may block for a long time, upon return the
|
||||
// selected set contains keys of the ready channels
|
||||
final int n = selector.select(100);
|
||||
|
||||
if (shouldEnableOpWrite)
|
||||
enableWrites();
|
||||
|
||||
checkThrottleInterval();
|
||||
|
||||
if (n == 0) {
|
||||
continue; // nothing to do
|
||||
}
|
||||
|
||||
final Set<SelectionKey> selectionKeys = selector.selectedKeys();
|
||||
for (final SelectionKey key : selectionKeys) {
|
||||
|
||||
try {
|
||||
|
||||
if (key.isReadable()) {
|
||||
final DatagramChannel socketChannel = (DatagramChannel) key.channel();
|
||||
reader.readAll(socketChannel);
|
||||
}
|
||||
|
||||
if (key.isWritable()) {
|
||||
final DatagramChannel socketChannel = (DatagramChannel) key.channel();
|
||||
try {
|
||||
int bytesJustWritten = writer.writeAll(socketChannel);
|
||||
contemplateThrottleWrites(bytesJustWritten);
|
||||
} catch (NotYetConnectedException e) {
|
||||
if (LOG.isDebugEnabled())
|
||||
LOG.debug("", e);
|
||||
serverConnector.connectLater();
|
||||
} catch (IOException e) {
|
||||
if (LOG.isDebugEnabled())
|
||||
LOG.debug("", e);
|
||||
serverConnector.connectLater();
|
||||
}
|
||||
}
|
||||
|
||||
} catch (Exception e) {
|
||||
LOG.error("", e);
|
||||
closeEarlyAndQuietly(key.channel());
|
||||
}
|
||||
}
|
||||
|
||||
selectionKeys.clear();
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
|
||||
private DatagramChannel connectClient() throws IOException {
|
||||
final DatagramChannel client = address.isMulticastAddress() ?
|
||||
DatagramChannel.open(address.getAddress().length == 4 ? INET : INET6) :
|
||||
DatagramChannel.open();
|
||||
|
||||
final InetSocketAddress hostAddress = new InetSocketAddress(port);
|
||||
client.configureBlocking(false);
|
||||
if (address.isMulticastAddress()) {
|
||||
client.setOption(IP_MULTICAST_IF, networkInterface);
|
||||
client.setOption(SO_REUSEADDR, true);
|
||||
client.bind(hostAddress);
|
||||
client.join(address, networkInterface);
|
||||
if (LOG.isDebugEnabled())
|
||||
LOG.debug("Connecting via multicast, group=" + address);
|
||||
} else {
|
||||
client.bind(hostAddress);
|
||||
}
|
||||
|
||||
if (LOG.isDebugEnabled())
|
||||
LOG.debug("Listening on port " + port);
|
||||
closeables.add(client);
|
||||
return client;
|
||||
}
|
||||
|
||||
/**
|
||||
* called whenever there is a change to the modification iterator
|
||||
*/
|
||||
@Override
|
||||
public void onChange() {
|
||||
// the write have to be enabled on the same thread as the selector
|
||||
shouldEnableOpWrite = true;
|
||||
selector.wakeup();
|
||||
}
|
||||
|
||||
private void enableWrites() {
|
||||
|
||||
if (writeChannel == null)
|
||||
return;
|
||||
|
||||
try {
|
||||
final SelectionKey selectionKey = writeChannel.keyFor(this.selector);
|
||||
if (selectionKey != null)
|
||||
selectionKey.interestOps(selectionKey.interestOps() | OP_WRITE);
|
||||
} catch (Exception e) {
|
||||
LOG.error("", e);
|
||||
}
|
||||
}
|
||||
|
||||
private void disableWrites() {
|
||||
|
||||
if (writeChannel == null)
|
||||
return;
|
||||
|
||||
try {
|
||||
final SelectionKey selectionKey = writeChannel.keyFor(this.selector);
|
||||
selectionKey.interestOps(selectionKey.interestOps() & ~OP_WRITE);
|
||||
} catch (Exception e) {
|
||||
LOG.error("", e);
|
||||
}
|
||||
}
|
||||
|
||||
private static class UdpSocketChannelEntryReader {
|
||||
private static class UdpSocketChannelEntryReader implements EntryReader {
|
||||
|
||||
private final Replica.EntryExternalizable externalizable;
|
||||
private final ByteBuffer in;
|
||||
@@ -246,7 +86,8 @@ class UdpReplicator extends AbstractChannelReplicator implements Replica.Modific
|
||||
* @throws IOException
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
void readAll(@NotNull final DatagramChannel socketChannel) throws IOException, InterruptedException {
|
||||
public void readAll(@NotNull final DatagramChannel socketChannel) throws IOException,
|
||||
InterruptedException {
|
||||
|
||||
out.clear();
|
||||
in.clear();
|
||||
@@ -276,20 +117,25 @@ class UdpReplicator extends AbstractChannelReplicator implements Replica.Modific
|
||||
|
||||
}
|
||||
|
||||
private class UdpSocketChannelEntryWriter {
|
||||
private static class UdpSocketChannelEntryWriter implements EntryWriter {
|
||||
|
||||
private final ByteBuffer out;
|
||||
private final ByteBufferBytes in;
|
||||
private final EntryCallback entryCallback;
|
||||
private final UdpChannelReplicator udpReplicator;
|
||||
private Replica.ModificationIterator modificationIterator;
|
||||
|
||||
UdpSocketChannelEntryWriter(final int serializedEntrySize,
|
||||
@NotNull final Replica.EntryExternalizable externalizable) {
|
||||
@NotNull final Replica.EntryExternalizable externalizable,
|
||||
@NotNull final Replica.ModificationIterator modificationIterator,
|
||||
UdpChannelReplicator udpReplicator) {
|
||||
this.udpReplicator = udpReplicator;
|
||||
|
||||
// we make the buffer twice as large just to give ourselves headroom
|
||||
out = ByteBuffer.allocateDirect(serializedEntrySize * 2);
|
||||
in = new ByteBufferBytes(out);
|
||||
entryCallback = new EntryCallback(externalizable, in);
|
||||
|
||||
this.modificationIterator = modificationIterator;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -308,7 +154,7 @@ class UdpReplicator extends AbstractChannelReplicator implements Replica.Modific
|
||||
* @throws InterruptedException
|
||||
* @throws IOException
|
||||
*/
|
||||
int writeAll(@NotNull final DatagramChannel socketChannel)
|
||||
public int writeAll(@NotNull final DatagramChannel socketChannel)
|
||||
throws InterruptedException, IOException {
|
||||
|
||||
out.clear();
|
||||
@@ -318,7 +164,7 @@ class UdpReplicator extends AbstractChannelReplicator implements Replica.Modific
|
||||
final boolean wasDataRead = modificationIterator.nextEntry(entryCallback, 0);
|
||||
|
||||
if (!wasDataRead) {
|
||||
disableWrites();
|
||||
udpReplicator.disableWrites();
|
||||
return 0;
|
||||
}
|
||||
|
||||
@@ -331,59 +177,6 @@ class UdpReplicator extends AbstractChannelReplicator implements Replica.Modific
|
||||
}
|
||||
}
|
||||
|
||||
private class ServerConnector extends AbstractChannelReplicator.AbstractConnector {
|
||||
private final InetSocketAddress socketAddress;
|
||||
|
||||
private ServerConnector() {
|
||||
super("UDP-Connector");
|
||||
this.socketAddress = new InetSocketAddress(address, port);
|
||||
}
|
||||
|
||||
SelectableChannel doConnect() throws
|
||||
IOException, InterruptedException {
|
||||
|
||||
final DatagramChannel server = DatagramChannel.open();
|
||||
server.configureBlocking(false);
|
||||
|
||||
// Kick off connection establishment
|
||||
try {
|
||||
// Create a non-blocking socket channel
|
||||
server.socket().setBroadcast(true);
|
||||
server.connect(socketAddress);
|
||||
} catch (IOException e) {
|
||||
if (LOG.isDebugEnabled())
|
||||
LOG.debug("details=" + new Details(socketAddress, localIdentifier), e);
|
||||
connectLater();
|
||||
return null;
|
||||
}
|
||||
|
||||
server.setOption(SO_REUSEADDR, true)
|
||||
.setOption(StandardSocketOptions.IP_MULTICAST_LOOP, false)
|
||||
.setOption(StandardSocketOptions.SO_BROADCAST, true)
|
||||
.setOption(SO_REUSEADDR, true);
|
||||
|
||||
// the registration has be be run on the same thread as the selector
|
||||
addPendingRegistration(new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
|
||||
try {
|
||||
server.register(selector, OP_WRITE);
|
||||
writeChannel = server;
|
||||
throttle(server);
|
||||
} catch (ClosedChannelException e) {
|
||||
LOG.error("", e);
|
||||
}
|
||||
|
||||
}
|
||||
});
|
||||
|
||||
return server;
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user