diff options
Diffstat (limited to 'java/src')
3 files changed, 135 insertions, 39 deletions
diff --git a/java/src/Ice/src/main/java/com/zeroc/Ice/ConnectionI.java b/java/src/Ice/src/main/java/com/zeroc/Ice/ConnectionI.java index ef086dbf88a..13b2d7a1dcc 100644 --- a/java/src/Ice/src/main/java/com/zeroc/Ice/ConnectionI.java +++ b/java/src/Ice/src/main/java/com/zeroc/Ice/ConnectionI.java @@ -165,25 +165,27 @@ public final class ConnectionI extends com.zeroc.IceInternal.EventHandler } @Override - synchronized public void close(boolean force) + synchronized public void close(ConnectionClose mode) { if(Thread.interrupted()) { throw new OperationInterruptedException(); } - if(force) + if(mode == ConnectionClose.CloseForcefully) { - setState(StateClosed, new ForcedCloseConnectionException()); + setState(StateClosed, new ConnectionManuallyClosedException(false)); + } + else if(mode == ConnectionClose.CloseGracefully) + { + setState(StateClosing, new ConnectionManuallyClosedException(true)); } else { + assert(mode == ConnectionClose.CloseGracefullyAndWait); + // - // If we do a graceful shutdown, then we wait until all - // outstanding requests have been completed. Otherwise, - // the CloseConnectionException will cause all outstanding - // requests to be retried, regardless of whether the - // server has processed them or not. + // Wait until all outstanding requests have been completed. // while(!_asyncRequests.isEmpty()) { @@ -197,7 +199,7 @@ public final class ConnectionI extends com.zeroc.IceInternal.EventHandler } } - setState(StateClosing, new CloseConnectionException()); + setState(StateClosing, new ConnectionManuallyClosedException(true)); } } @@ -289,14 +291,14 @@ public final class ConnectionI extends com.zeroc.IceInternal.EventHandler // We send a heartbeat if there was no activity in the last // (timeout / 4) period. Sending a heartbeat sooner than // really needed is safer to ensure that the receiver will - // receive in time the heartbeat. Sending the heartbeat if + // receive the heartbeat in time. Sending the heartbeat if // there was no activity in the last (timeout / 2) period // isn't enough since monitor() is called only every (timeout // / 2) period. // // Note that this doesn't imply that we are sending 4 // heartbeats per timeout period because the monitor() method - // is sill only called every (timeout / 2) period. + // is still only called every (timeout / 2) period. // if(acm.heartbeat == ACMHeartbeat.HeartbeatAlways || (acm.heartbeat != ACMHeartbeat.HeartbeatOff && _writeStream.isEmpty() && @@ -304,7 +306,7 @@ public final class ConnectionI extends com.zeroc.IceInternal.EventHandler { if(acm.heartbeat != ACMHeartbeat.HeartbeatOnInvocation || _dispatchCount > 0) { - heartbeat(); + sendHeartbeatNow(); } } @@ -480,6 +482,107 @@ public final class ConnectionI extends com.zeroc.IceInternal.EventHandler } @Override + public void heartbeat() + { + ObjectPrx.waitForResponseForCompletion(heartbeatAsync()); + } + + private class HeartbeatAsync extends com.zeroc.IceInternal.OutgoingAsyncBaseI<Void> + { + public HeartbeatAsync(Communicator communicator, com.zeroc.IceInternal.Instance instance) + { + super(communicator, instance, "heartbeat"); + } + + @Override + public Connection getConnection() + { + return ConnectionI.this; + } + + @Override + protected void markSent() + { + super.markSent(); + + assert((_state & StateOK) != 0); + complete(null); + } + + @Override + protected void markCompleted() + { + if(_exception != null) + { + completeExceptionally(_exception); + } + super.markCompleted(); + } + + public void invoke() + { + try + { + _os.writeBlob(Protocol.magic); + ProtocolVersion.ice_write(_os, Protocol.currentProtocol); + EncodingVersion.ice_write(_os, Protocol.currentProtocolEncoding); + _os.writeByte(Protocol.validateConnectionMsg); + _os.writeByte((byte) 0); + _os.writeInt(Protocol.headerSize); // Message size. + + int status; + if(_instance.queueRequests()) + { + status = _instance.getQueueExecutor().execute(new Callable<Integer>() + { + @Override + public Integer call() + throws com.zeroc.IceInternal.RetryException + { + return ConnectionI.this.sendAsyncRequest(HeartbeatAsync.this, false, false, 0); + } + }); + } + else + { + status = ConnectionI.this.sendAsyncRequest(this, false, false, 0); + } + + if((status & AsyncStatus.Sent) > 0) + { + _sentSynchronously = true; + if((status & AsyncStatus.InvokeSentCallback) > 0) + { + invokeSent(); + } + } + } + catch(com.zeroc.IceInternal.RetryException ex) + { + if(completed(ex.get())) + { + invokeCompletedAsync(); + } + } + catch(com.zeroc.Ice.Exception ex) + { + if(completed(ex)) + { + invokeCompletedAsync(); + } + } + } + } + + @Override + public java.util.concurrent.CompletableFuture<Void> heartbeatAsync() + { + HeartbeatAsync __f = new HeartbeatAsync(_communicator, _instance); + __f.invoke(); + return __f; + } + + @Override synchronized public void setACM(java.util.OptionalInt timeout, java.util.Optional<ACMClose> close, java.util.Optional<ACMHeartbeat> heartbeat) { @@ -664,8 +767,7 @@ public final class ConnectionI extends com.zeroc.IceInternal.EventHandler public synchronized void invokeException(int requestId, LocalException ex, int invokeNum, boolean amd) { // - // Fatal exception while invoking a request. Since - // sendResponse/sendNoResponse isn't + // Fatal exception while invoking a request. Since sendResponse/sendNoResponse isn't // called in case of a fatal exception we decrement _dispatchCount here. // @@ -931,7 +1033,7 @@ public final class ConnectionI extends com.zeroc.IceInternal.EventHandler } else { - assert (_state <= StateClosingPending); + assert(_state <= StateClosingPending); // // We parse messages first, if we receive a close @@ -1107,7 +1209,8 @@ public final class ConnectionI extends com.zeroc.IceInternal.EventHandler // if(info.invokeNum > 0) { - invokeAll(info.stream, info.invokeNum, info.requestId, info.compress, info.servantManager, info.adapter); + invokeAll(info.stream, info.invokeNum, info.requestId, info.compress, info.servantManager, + info.adapter); // // Don't increase dispatchedCount, the dispatch count is @@ -1264,7 +1367,7 @@ public final class ConnectionI extends com.zeroc.IceInternal.EventHandler // Trace the cause of unexpected connection closures // if(!(_exception instanceof CloseConnectionException || - _exception instanceof ForcedCloseConnectionException || + _exception instanceof ConnectionManuallyClosedException || _exception instanceof ConnectionTimeoutException || _exception instanceof CommunicatorDestroyedException || _exception instanceof ObjectAdapterDeactivatedException)) @@ -1614,7 +1717,7 @@ public final class ConnectionI extends com.zeroc.IceInternal.EventHandler // Don't warn about certain expected exceptions. // if(!(_exception instanceof CloseConnectionException || - _exception instanceof ForcedCloseConnectionException || + _exception instanceof ConnectionManuallyClosedException || _exception instanceof ConnectionTimeoutException || _exception instanceof CommunicatorDestroyedException || _exception instanceof ObjectAdapterDeactivatedException || @@ -1804,7 +1907,7 @@ public final class ConnectionI extends com.zeroc.IceInternal.EventHandler if(_observer != null && state == StateClosed && _exception != null) { if(!(_exception instanceof CloseConnectionException || - _exception instanceof ForcedCloseConnectionException || + _exception instanceof ConnectionManuallyClosedException || _exception instanceof ConnectionTimeoutException || _exception instanceof CommunicatorDestroyedException || _exception instanceof ObjectAdapterDeactivatedException || @@ -1833,8 +1936,7 @@ public final class ConnectionI extends com.zeroc.IceInternal.EventHandler private void initiateShutdown() { - assert (_state == StateClosing); - assert (_dispatchCount == 0); + assert(_state == StateClosing && _dispatchCount == 0); if(_shutdownInitiated) { @@ -1861,8 +1963,7 @@ public final class ConnectionI extends com.zeroc.IceInternal.EventHandler setState(StateClosingPending); // - // Notify the the transceiver of the graceful connection - // closure. + // Notify the transceiver of the graceful connection closure. // int op = _transceiver.closing(true, _exception); if(op != 0) @@ -1874,7 +1975,7 @@ public final class ConnectionI extends com.zeroc.IceInternal.EventHandler } } - private void heartbeat() + private void sendHeartbeatNow() { assert (_state == StateActive); @@ -2071,8 +2172,7 @@ public final class ConnectionI extends com.zeroc.IceInternal.EventHandler } else if(_state == StateClosingPending && _writeStream.pos() == 0) { - // Message wasn't sent, empty the _writeStream, we're not going to - // send more data. + // Message wasn't sent, empty the _writeStream, we're not going to send more data. OutgoingMessage message = _sendStreams.getFirst(); _writeStream.swap(message.stream); return SocketOperation.None; @@ -2149,9 +2249,8 @@ public final class ConnectionI extends com.zeroc.IceInternal.EventHandler } // - // If all the messages were sent and we are in the closing state, we - // schedule the close timeout to wait for the peer to close the - // connection. + // If all the messages were sent and we are in the closing state, we schedule + // the close timeout to wait for the peer to close the connection. // if(_state == StateClosing && _shutdownInitiated) { @@ -2376,8 +2475,7 @@ public final class ConnectionI extends com.zeroc.IceInternal.EventHandler setState(StateClosingPending, new CloseConnectionException()); // - // Notify the the transceiver of the graceful connection - // closure. + // Notify the transceiver of the graceful connection closure. // int op = _transceiver.closing(false, _exception); if(op != 0) @@ -2393,9 +2491,8 @@ public final class ConnectionI extends com.zeroc.IceInternal.EventHandler { if(_state >= StateClosing) { - TraceUtil.trace("received request during closing\n" - + "(ignored by server, client will retry)", info.stream, _logger, - _traceLevels); + TraceUtil.trace("received request during closing\n(ignored by server, client will retry)", + info.stream, _logger, _traceLevels); } else { @@ -2413,9 +2510,8 @@ public final class ConnectionI extends com.zeroc.IceInternal.EventHandler { if(_state >= StateClosing) { - TraceUtil.trace("received batch request during closing\n" - + "(ignored by server, client will retry)", info.stream, _logger, - _traceLevels); + TraceUtil.trace("received batch request during closing\n(ignored by server, client will retry)", + info.stream, _logger, _traceLevels); } else { diff --git a/java/src/Ice/src/main/java/com/zeroc/IceInternal/IncomingConnectionFactory.java b/java/src/Ice/src/main/java/com/zeroc/IceInternal/IncomingConnectionFactory.java index 337e56f05f7..98dd813fb66 100644 --- a/java/src/Ice/src/main/java/com/zeroc/IceInternal/IncomingConnectionFactory.java +++ b/java/src/Ice/src/main/java/com/zeroc/IceInternal/IncomingConnectionFactory.java @@ -137,7 +137,7 @@ public final class IncomingConnectionFactory extends EventHandler implements Con // for(ConnectionI c : connections) { - c.close(true); + c.close(com.zeroc.Ice.ConnectionClose.CloseForcefully); } throw e; } diff --git a/java/src/Ice/src/main/java/com/zeroc/IceInternal/OutgoingConnectionFactory.java b/java/src/Ice/src/main/java/com/zeroc/IceInternal/OutgoingConnectionFactory.java index 3e62623a17f..9bd778ecfb4 100644 --- a/java/src/Ice/src/main/java/com/zeroc/IceInternal/OutgoingConnectionFactory.java +++ b/java/src/Ice/src/main/java/com/zeroc/IceInternal/OutgoingConnectionFactory.java @@ -138,7 +138,7 @@ public final class OutgoingConnectionFactory { for(ConnectionI c : l) { - c.close(true); + c.close(com.zeroc.Ice.ConnectionClose.CloseForcefully); } } throw new com.zeroc.Ice.OperationInterruptedException(); |