Add on connecting and dtls retransmission.

Signed-off-by: Achim Kraus <[email protected]>

Author: Achim Kraus

californium-core/src/main/java/org/eclipse/californium/core/coap/Message.java

@@ -38,6 +38,7 @@
  *    Achim Kraus (Bosch Software Innovations GmbH) - move onContextEstablished
  *                                                    to MessageObserver.
  *                                                    Issue #487
+ *    Achim Kraus (Bosch Software Innovations GmbH) - add onConnect
  ******************************************************************************/
 package org.eclipse.californium.core.coap;
 
@@ -707,6 +708,28 @@ public void setReadyToSend() {
 		}
 	}
 
+	/**
+	 * Indicate, that this message triggered the connector to establish a
+	 * connection. Not part of the fluent API.
+	 */
+	public void onConnecting() {
+		for (MessageObserver handler : getMessageObservers()) {
+			handler.onConnecting();
+		}
+	}
+
+	/**
+	 * Indicate, that this message triggered the connector to establish a
+	 * connection and a dtls handshake flight was retransmitted.
+	 * 
+	 * @param flight {@code 1 ... 6}, number of retransmitted flight.
+	 */
+	public void onDtlsRetransmission(int flight) {
+		for (MessageObserver handler : getMessageObservers()) {
+			handler.onDtlsRetransmission(flight);
+		}
+	}
+
 	/**
 	 * Checks if this message has been sent.
 	 * 

californium-core/src/main/java/org/eclipse/californium/core/coap/MessageObserver.java

@@ -25,6 +25,7 @@
  *    Achim Kraus (Bosch Software Innovations GmbH) - move onContextEstablished
  *                                                    to MessageObserver.
  *                                                    Issue #487
+ *    Achim Kraus (Bosch Software Innovations GmbH) - add onConnect
  ******************************************************************************/
 package org.eclipse.californium.core.coap;
 
@@ -42,6 +43,10 @@
  * still has not received anything from the remote endpoint</li>
  * <li>{@link #onCancel()} when the message has been canceled</li>
  * <li>{@link #onReadyToSend()} right before the message is being sent</li>
+ * <li>{@link #onConnecting()} right before a connector establish a connection. 
+ * Not called, if the connection is already established or the connector doesn't
+ * require to establish a connection.</li>
+ * <li>{@link #onDtlsRetransmission()} when a dtls handshake flight is retransmitted.</li>
  * <li>{@link #onSent()} right after the message has been sent
  * (successfully)</li>
  * <li>{@link #onSendError(Throwable)} if the message cannot be sent</li>
@@ -109,6 +114,20 @@ public interface MessageObserver {
 	 */
 	void onReadyToSend();
 
+	/**
+	 * Invoked, when connector requires to establish a connection before sending
+	 * the message.
+	 */
+	void onConnecting();
+
+	/**
+	 * Indicate, that this message triggered the connector to establish a
+	 * connection and a dtls handshake flight was retransmitted.
+	 * 
+	 * @param flight {@code 1 ... 6}, number of retransmitted flight.
+	 */
+	void onDtlsRetransmission(int flight);
+
 	/**
 	 * Invoked right after the message has been sent.
 	 * <p>

californium-core/src/main/java/org/eclipse/californium/core/coap/MessageObserverAdapter.java

@@ -27,6 +27,7 @@
  *    Achim Kraus (Bosch Software Innovations GmbH) - move onContextEstablished
  *                                                    to MessageObserver.
  *                                                    Issue #487
+ *    Achim Kraus (Bosch Software Innovations GmbH) - add onConnect
  ******************************************************************************/
 package org.eclipse.californium.core.coap;
 
@@ -83,6 +84,16 @@ public void onReadyToSend() {
 		// empty default implementation
 	}
 
+	@Override
+	public void onConnecting() {
+		// empty default implementation
+	}
+
+	@Override
+	public void onDtlsRetransmission(int flight) {
+		// empty default implementation
+	}
+
 	@Override
 	public void onSent() {
 		// empty default implementation

californium-core/src/main/java/org/eclipse/californium/core/network/CoapEndpoint.java

@@ -58,6 +58,7 @@
  *    Achim Kraus (Bosch Software Innovations GmbH) - reject messages with MID only
  *                                                    (therefore tcp messages are not rejected)
  *    Achim Kraus (Bosch Software Innovations GmbH) - setup retransmitResponse for notifies
+ *    Achim Kraus (Bosch Software Innovations GmbH) - forward onConnect
  ******************************************************************************/
 package org.eclipse.californium.core.network;
 
@@ -1030,6 +1031,16 @@ public SendingCallback(final Message message) {
 			this.message = message;
 		}
 
+		@Override
+		public void onConnecting() {
+			message.onConnecting();
+		}
+
+		@Override
+		public void onDtlsRetransmission(int flight) {
+			message.onDtlsRetransmission(flight);
+		}
+
 		@Override
 		public void onContextEstablished(EndpointContext context) {
 

element-connector-tcp/src/main/java/org/eclipse/californium/elements/tcp/TcpClientConnector.java

@@ -32,6 +32,7 @@
  * Bosch Software Innovations GmbH - migrate to SLF4J
  * Achim Kraus (Bosch Software Innovations GmbH) - add logs for create and close channel
  * Achim Kraus (Bosch Software Innovations GmbH) - adjust logging
+ * Achim Kraus (Bosch Software Innovations GmbH) - add onConnect
  ******************************************************************************/
 package org.eclipse.californium.elements.tcp;
 
@@ -145,14 +146,17 @@ public void destroy() {
 	@Override
 	public void send(final RawData msg) {
 		InetSocketAddress addressKey = new InetSocketAddress(msg.getAddress(), msg.getPort());
+		boolean connected = poolMap.contains(addressKey);
 		final EndpointContextMatcher endpointMatcher = getEndpointContextMatcher();
 		/* check, if a new connection should be established */
-		if (endpointMatcher != null && !poolMap.contains(addressKey)
-				&& !endpointMatcher.isToBeSent(msg.getEndpointContext(), null)) {
+		if (endpointMatcher != null && !connected && !endpointMatcher.isToBeSent(msg.getEndpointContext(), null)) {
 			LOGGER.warn("TcpConnector drops {} bytes to new {}:{}", msg.getSize(), msg.getAddress(), msg.getPort());
 			msg.onError(new EndpointMismatchException("no connection"));
 			return;
 		}
+		if (!connected) {
+			msg.onConnecting();
+		}
 		final ChannelPool channelPool = poolMap.get(addressKey);
 		Future<Channel> acquire = channelPool.acquire();
 		acquire.addListener(new GenericFutureListener<Future<Channel>>() {

element-connector-tcp/src/test/java/org/eclipse/californium/elements/tcp/TcpConnectorTest.java

@@ -33,6 +33,7 @@
 
 import org.eclipse.californium.elements.Connector;
 import org.eclipse.californium.elements.RawData;
+import org.eclipse.californium.elements.util.SimpleMessageCallback;
 import org.junit.After;
 import org.junit.Rule;
 import org.junit.Test;
@@ -150,4 +151,37 @@ public void singleServerManyClients() throws Exception {
 			assertTrue("Received unexpected message: " + received, matched);
 		}
 	}
+
+	@Test
+	public void onConnect() throws Exception {
+		TcpServerConnector server = new TcpServerConnector(createServerAddress(0), NUMBER_OF_THREADS,
+				IDLE_TIMEOUT_IN_S);
+		TcpClientConnector client = new TcpClientConnector(NUMBER_OF_THREADS, CONNECTION_TIMEOUT_IN_MS,
+				IDLE_TIMEOUT_IN_S);
+
+		cleanup.add(server);
+		cleanup.add(client);
+
+		Catcher serverCatcher = new Catcher();
+		Catcher clientCatcher = new Catcher();
+		server.setRawDataReceiver(serverCatcher);
+		client.setRawDataReceiver(clientCatcher);
+		server.start();
+		client.start();
+
+		SimpleMessageCallback callback = new SimpleMessageCallback();
+		RawData msg = createMessage(server.getAddress(), messageSize, callback);
+
+		client.send(msg);
+		serverCatcher.blockUntilSize(1);
+		assertTrue(callback.isConnecting());
+
+		callback = new SimpleMessageCallback();
+		msg = createMessage(server.getAddress(), messageSize, callback);
+
+		client.send(msg);
+		serverCatcher.blockUntilSize(2);
+		assertFalse(callback.isConnecting());
+
+	}
 }

element-connector/src/main/java/org/eclipse/californium/elements/MessageCallback.java

@@ -19,11 +19,26 @@
  *                                                    issue #305
  *    Achim Kraus (Bosch Software Innovations GmbH) - add comment on processing
  *                                                    onContextEstablished. issue #311 
+ *    Achim Kraus (Bosch Software Innovations GmbH) - add onConnect
  ******************************************************************************/
 package org.eclipse.californium.elements;
 
 public interface MessageCallback {
 
+	/**
+	 * Called, when connector requires to establish a connection. Not called, if
+	 * the connection is already established or the connector doesn't require to
+	 * establish a connection.
+	 */
+	void onConnecting();
+
+	/**
+	 * Called, when the dtls connector retransmits a handshake flight.
+	 * 
+	 * @param flight {@code 1 ... 6}, number of retransmitted flight.
+	 */
+	void onDtlsRetransmission(int flight);
+
 	/**
 	 * Called when the context information for an outbound message has been
 	 * established.

element-connector/src/main/java/org/eclipse/californium/elements/RawData.java

@@ -27,6 +27,7 @@
  *                                                    constructors.
  *    Achim Kraus (Bosch Software Innovations GmbH) - replace isSecure by 
  *                                                    connector's protocol
+ *    Achim Kraus (Bosch Software Innovations GmbH) - add onConnect
  ******************************************************************************/
 package org.eclipse.californium.elements;
 
@@ -56,18 +57,18 @@ public final class RawData {
 	public final byte[] bytes;
 
 	/** Indicates if this message is a multicast message */
-	private boolean multicast;
+	private final boolean multicast;
 
 	/**
 	 * Endpoint context of the remote peer.
 	 */
-	private EndpointContext peerEndpointContext;
+	private final EndpointContext peerEndpointContext;
 
 	/**
 	 * Message callback to receive the actual endpoint context the message is
 	 * sent in.
 	 */
-	private MessageCallback callback;
+	private final MessageCallback callback;
 
 	/**
 	 * Instantiates a new raw data.
@@ -221,6 +222,28 @@ public EndpointContext getEndpointContext() {
 		return peerEndpointContext;
 	}
 
+	/**
+	 * Callback, when connector requires to establish a connection. Not called,
+	 * if the connection is already established or the connector doesn't require
+	 * to establish a connection.
+	 */
+	public void onConnecting() {
+		if (null != callback) {
+			callback.onConnecting();
+		}
+	}
+
+	/**
+	 * Callback, when the dtls connector retransmits a handshake flight.
+	 * 
+	 * @param flight {@code 1 ... 6}, number of retransmitted flight.
+	 */
+	public void onDtlsRetransmission(int flight) {
+		if (null != callback) {
+			callback.onDtlsRetransmission(flight);
+		}
+	}
+
 	/**
 	 * Callback, when context gets available. Used on sending an message.
 	 * 

element-connector/src/test/java/org/eclipse/californium/elements/util/SimpleMessageCallback.java

@@ -56,6 +56,14 @@ public class SimpleMessageCallback implements MessageCallback {
 	 * Indicator for message sent.
 	 */
 	private boolean sent;
+	/**
+	 * Indicator for connect.
+	 */
+	private boolean connecting;
+	/**
+	 * Retransmitted dtls handshake flight.
+	 */
+	private int retransmittedDtlsFlight;
 
 	/**
 	 * Create new message callback.
@@ -92,6 +100,16 @@ public SimpleMessageCallback(int calls, boolean countContextEstablished, Message
 		this.chained = chained;
 	}
 
+	@Override
+	public synchronized void onConnecting() {
+		connecting = true;
+	}
+
+	@Override
+	public synchronized void onDtlsRetransmission(int flight) {
+		retransmittedDtlsFlight = flight;
+	}
+
 	@Override
 	public synchronized void onContextEstablished(EndpointContext context) {
 		this.context = context;
@@ -179,6 +197,24 @@ public synchronized boolean isSent() {
 		return sent;
 	}
 
+	/**
+	 * Check, if message requires to establish a connection.
+	 * 
+	 * @return {@code true}, if requires a connection, {@code false}, otherwise
+	 */
+	public synchronized boolean isConnecting() {
+		return connecting;
+	}
+
+	/**
+	 * Check, if message requires to establish a connection.
+	 * 
+	 * @return {@code true}, if requires a connection, {@code false}, otherwise
+	 */
+	public synchronized int getRetransmittedDtlsFlight() {
+		return retransmittedDtlsFlight;
+	}
+
 	/**
 	 * Get error of sending message.
 	 * 

scandium-core/src/main/java/org/eclipse/californium/scandium/DTLSConnector.java

@@ -91,6 +91,7 @@
  *                                                    Move session listener callback out of sync
  *                                                    block of processApplicationDataRecord.
  *    Achim Kraus (Bosch Software Innovations GmbH) - add handshakeFlightRetransmitted
+ *    Achim Kraus (Bosch Software Innovations GmbH) - add onConnect
  ******************************************************************************/
 package org.eclipse.californium.scandium;
 
@@ -1462,8 +1463,9 @@ private void sendMessage(final RawData message) throws HandshakeException {
 			if (!checkOutboundEndpointContext(message, null)) {
 				return;
 			}
-			// no session with peer established yet, create new empty session &
-			// start handshake
+			message.onConnecting();
+			// no session with peer established yet, 
+			// create new empty session & start handshake
 			Handshaker handshaker = new ClientHandshaker(
 					DTLSSession.newClientSession(peerAddress, message.getEndpointContext().getVirtualHost()),
 					getRecordLayerForPeer(connection),
@@ -1477,6 +1479,7 @@ private void sendMessage(final RawData message) throws HandshakeException {
 		// TODO what if there already is an ongoing handshake with the peer
 		else if (connection.isResumptionRequired()){
 			// create the session to resume from the previous one.
+			message.onConnecting();
 			SessionId sessionId;
 			if (ticket == null) {
 				ticket = session.getSessionTicket();
@@ -1562,7 +1565,13 @@ public void sessionEstablished(Handshaker handshaker, DTLSSession establishedSes
 				LOGGER.debug("Session with [{}] established, now sending deferred message", establishedSession.getPeer());
 				sendMessage(message, establishedSession);
 			}
-			
+
+			@Override
+			public void handshakeFlightRetransmitted(Handshaker handshaker, int flight) {
+				LOGGER.debug("Session with [{}] retransmit flight {}", handshaker.getPeerAddress(), flight);
+				message.onDtlsRetransmission(flight);
+			}
+
 			@Override
 			public void handshakeFailed(Handshaker handshaker, Throwable error) {
 				LOGGER.debug("Session with [{}] failed, report error", handshaker.getPeerAddress());