Merge pull request 'Possible solution for #23249: prevent grey-out because of collision between pings and packets' (#12) from feature/23249 into develop

Author: fmichielssen

src/main/java/eu/openanalytics/containerproxy/service/HeartbeatService.java

@@ -44,6 +44,7 @@
 import eu.openanalytics.containerproxy.model.runtime.ProxyStatus;
 import eu.openanalytics.containerproxy.util.DelegatingStreamSinkConduit;
 import eu.openanalytics.containerproxy.util.DelegatingStreamSourceConduit;
+import eu.openanalytics.containerproxy.util.ChannelActiveListener;
 import io.undertow.server.HttpServerExchange;
 import io.undertow.server.protocol.http.HttpServerConnection;
 
@@ -127,27 +128,38 @@ private void wrapChannels(StreamConnection streamConn) {
 			if (!streamConn.isOpen()) return;
 
 			ConduitStreamSinkChannel sinkChannel = streamConn.getSinkChannel();
-			DelegatingStreamSinkConduit conduitWrapper = new DelegatingStreamSinkConduit(sinkChannel.getConduit(), null);
+			ChannelActiveListener writeListener = new ChannelActiveListener();
+			DelegatingStreamSinkConduit conduitWrapper = new DelegatingStreamSinkConduit(sinkChannel.getConduit(), writeListener);
 			sinkChannel.setConduit(conduitWrapper);
 
 			ConduitStreamSourceChannel sourceChannel = streamConn.getSourceChannel();
 			DelegatingStreamSourceConduit srcConduitWrapper = new DelegatingStreamSourceConduit(sourceChannel.getConduit(), data -> checkPong(data));
 			sourceChannel.setConduit(srcConduitWrapper);
 
-			heartbeatExecutor.schedule(() -> sendPing(streamConn), getHeartbeatRate(), TimeUnit.MILLISECONDS);
+			heartbeatExecutor.schedule(() -> sendPing(writeListener, streamConn), getHeartbeatRate(), TimeUnit.MILLISECONDS);
 		}
 
-		private void sendPing(StreamConnection streamConn) {
+		private void sendPing(ChannelActiveListener writeListener, StreamConnection streamConn) {
+			if (writeListener.isActive(getHeartbeatRate())) {
+				// active means that data was written to the channel in the least heartbeat interval
+				// therefore we don't send a ping now to not cause collisions
+
+				// reschedule ping
+				heartbeatExecutor.schedule(() -> sendPing(writeListener, streamConn), getHeartbeatRate(), TimeUnit.MILLISECONDS);
+				// mark as we received a heartbeat
+				heartbeatReceived(proxyId);
+				return;
+			}
 			if (!streamConn.isOpen()) return;
 
 			try {
-				streamConn.getSinkChannel().write(ByteBuffer.wrap(WEBSOCKET_PING));
+				((DelegatingStreamSinkConduit) streamConn.getSinkChannel().getConduit()).writeWithoutNotifying(ByteBuffer.wrap(WEBSOCKET_PING));
 				streamConn.getSinkChannel().flush();
 			} catch (IOException e) {
 				// Ignore failure, keep trying as long as the stream connection is valid.
 			}
 
-			heartbeatExecutor.schedule(() -> sendPing(streamConn), getHeartbeatRate(), TimeUnit.MILLISECONDS);
+			heartbeatExecutor.schedule(() -> sendPing(writeListener, streamConn), getHeartbeatRate(), TimeUnit.MILLISECONDS);
 		}
 
 		private void checkPong(byte[] response) {

src/main/java/eu/openanalytics/containerproxy/util/ChannelActiveListener.java

@@ -0,0 +1,54 @@
+/**
+ * ContainerProxy
+ *
+ * Copyright (C) 2016-2020 Open Analytics
+ *
+ * ===========================================================================
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Apache License as published by
+ * The Apache Software Foundation, either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * Apache License for more details.
+ *
+ * You should have received a copy of the Apache License
+ * along with this program.  If not, see <http://www.apache.org/licenses/>
+ */
+package eu.openanalytics.containerproxy.util;
+
+/**
+ * A listener that keeps track of whether a channel is active.
+ *
+ */
+public class ChannelActiveListener implements Runnable {
+
+	private long lastWrite = 0;
+	
+	@Override
+	public void run() {
+		lastWrite = System.currentTimeMillis();
+	}
+	
+	/**
+	 * Checks whether the channel was active in the provided period.
+	 */
+	public boolean isActive(long period) {
+		long diff = System.currentTimeMillis() - lastWrite;
+		
+		// make sure the period is at least 5 seconds
+		// this ensures that when the socket is active, the ping is delayed for at least 5 seconds
+		if (period < 5000) {
+			period = 5000;
+		}
+
+		if (diff <= period) {
+			return true;
+		}
+		return false;
+	}
+
+}

src/main/java/eu/openanalytics/containerproxy/util/DelegatingStreamSinkConduit.java

@@ -35,9 +35,11 @@
 public class DelegatingStreamSinkConduit implements StreamSinkConduit {
 
 	private StreamSinkConduit delegate;
-	private Consumer<byte[]> writeListener;
+	private Runnable writeListener;
 
-	public DelegatingStreamSinkConduit(StreamSinkConduit delegate, Consumer<byte[]> writeListener) {
+
+	
+	public DelegatingStreamSinkConduit(StreamSinkConduit delegate, Runnable writeListener) {
 		this.delegate = delegate;
 		this.writeListener = writeListener;
 	}
@@ -119,14 +121,14 @@ public long transferFrom(StreamSourceChannel source, long count, ByteBuffer thro
 
 	@Override
 	public int write(ByteBuffer src) throws IOException {
-		if (writeListener == null) {
-			return delegate.write(src);
-		} else {
-			byte[] data = new byte[src.remaining()];
-			src.get(data);
-			writeListener.accept(data);
-			return delegate.write(ByteBuffer.wrap(data));
+		if (writeListener != null) {
+			writeListener.run();
 		}
+		return delegate.write(src);
+	}
+
+	public int writeWithoutNotifying(ByteBuffer src) throws IOException {
+		return delegate.write(src);
 	}
 
 	@Override