diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 281b3708c..9f34415c6 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -694,9 +694,9 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo attached.hasRemoteHeartbeatInterval = true; // now we're finished we can get on with reading the entries - attached.handShakingComplete = true; attached.remoteModificationIterator.dirtyEntries(attached.remoteBootstrapTimestamp); reader.entriesFromBuffer(attached, key); + attached.handShakingComplete = true; } } @@ -738,10 +738,18 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo if (!entryWriter.hasBytesToWrite() && !entryWriter.isWorkIncomplete() - && hasNext(attached)) + && !hasNext(attached) + && attached.isHandShakingComplete()) { // if we have no more data to write to the socket then we will // un-register OP_WRITE on the selector, until more data becomes available + // The OP_WRITE will get added back in as soon as we have data to write + + if (LOG.isDebugEnabled()) + LOG.debug("Disabling OP_WRITE to remoteIdentifier=" + + attached.remoteIdentifier + + ", localIdentifier=" + localIdentifier); key.interestOps(key.interestOps() & ~OP_WRITE); + } } catch (IOException e) { quietClose(key, e); if (!attached.isServer) @@ -752,7 +760,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo private boolean hasNext(Attached attached) { return attached.remoteModificationIterator != null - && !attached.remoteModificationIterator.hasNext(); + && attached.remoteModificationIterator.hasNext(); } /** @@ -891,6 +899,9 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo final ServerSocketChannel serverChannel = openServerSocketChannel(); + if (serverChannel == null) + return null; + serverChannel.socket().setReceiveBufferSize(BUFFER_SIZE); serverChannel.configureBlocking(false); serverChannel.register(TcpReplicator.this.selector, 0); @@ -903,8 +914,10 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo closeables.add(serverSocket); } - serverSocket.setReuseAddress(true); + if (serverSocket == null) + return null; + serverSocket.setReuseAddress(true); serverSocket.bind(details.address()); // these can be run on this thread @@ -1014,6 +1027,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo @Nullable public Replica.ModificationIterator remoteModificationIterator; + // used for connect later public AbstractConnector connector; public long remoteBootstrapTimestamp = Long.MIN_VALUE; public byte remoteIdentifier = Byte.MIN_VALUE; @@ -1138,12 +1152,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo void entriesToBuffer(@NotNull final Replica.ModificationIterator modificationIterator, @NotNull final SelectionKey selectionKey) throws InterruptedException { - final SocketChannel socketChannel = (SocketChannel) selectionKey.channel(); - final Attached attached = (Attached) selectionKey.attachment(); - - // this can occur when new map are added to a channel - final boolean handShakingComplete = attached.isHandShakingComplete(); - int entriesWritten = 0; try { for (; ; entriesWritten++) { @@ -1237,30 +1245,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo in().writeLong(localHeartbeatInterval); } - /** - * removes back in the OP_WRITE from the selector, otherwise it'll spin loop. The OP_WRITE - * will get added back in as soon as we have data to write - * - * @param socketChannel the socketChannel we wish to stop writing to - * @param attached data associated with the socketChannels key - */ - public synchronized void disableWrite(@NotNull final SocketChannel socketChannel, - @NotNull final Attached attached) { - try { - SelectionKey key = socketChannel.keyFor(selector); - if (key != null) { - if (attached.isHandShakingComplete() && selector.isOpen()) { - if (LOG.isDebugEnabled()) - LOG.debug("Disabling OP_WRITE to remoteIdentifier=" + - attached.remoteIdentifier + - ", localIdentifier=" + localIdentifier); - key.interestOps(key.interestOps() & ~OP_WRITE); - } - } - } catch (Exception e) { - LOG.error("", e); - } - } public boolean doWork() { return uncompletedWork != null && uncompletedWork.doWork(in());