Generally processing of exception with serial execution.

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

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

@@ -67,6 +67,9 @@
  *                                                    was started.
  *    Achim Kraus (Bosch Software Innovations GmbH) - cancel pending messages on stop().
  *    Achim Kraus (Bosch Software Innovations GmbH) - add support for multicast
+ *    Achim Kraus (Bosch Software Innovations GmbH) - move response retransmission
+ *                                                    setup to BaseCoapStack to include
+ *                                                    it also in a try-catch
  ******************************************************************************/
 package org.eclipse.californium.core.network;
 
@@ -547,12 +550,6 @@ public void sendResponse(final Exchange exchange, final Response response) {
 			exchange.execute(new Runnable() {
 				@Override
 				public void run() {
-					if (exchange.getRequest().getOptions().hasObserve()) {
-						// observe- or cancel-observe-requests may have multiple responses
-						// when observes are finished, the last response has no longer an
-						// observe option. Therefore check the request for it.
-						exchange.retransmitResponse();
-					}
 					coapstack.sendResponse(exchange, response);
 				}
 			});

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

@@ -233,7 +233,12 @@ public void receiveRequest(final Request request, final EndpointReceiver receive
 
 				@Override
 				public void run() {
-					receiver.receiveRequest(previous, request);
+					try {
+						receiver.receiveRequest(previous, request);
+					} catch (RuntimeException ex) {
+						LOGGER.warn("error receiving request {} again!", request, ex);
+						receiver.reject(request);
+					}
 				}
 			});
 		} else {
@@ -242,7 +247,12 @@ public void run() {
 
 				@Override
 				public void run() {
-					receiver.receiveRequest(exchange, request);
+					try {
+						receiver.receiveRequest(exchange, request);
+					} catch (RuntimeException ex) {
+						LOGGER.warn("error receiving request {}", request, ex);
+						receiver.reject(request);
+					}
 				}
 			});
 		}
@@ -294,8 +304,8 @@ public void run() {
 									receiver.receiveResponse(prev, response);
 									return;
 								}
-							} catch (Exception ex) {
-								LOGGER.error("error receiving response {} for {}", response, prev, ex);
+							} catch (RuntimeException ex) {
+								LOGGER.warn("error receiving response {} for {}", response, prev, ex);
 							}
 							reject(response, receiver);
 						}
@@ -388,8 +398,8 @@ public void run() {
 						LOGGER.debug("ignoring potentially forged response for token {} with non-matching endpoint context",
 								idByToken);
 					}
-				} catch (Exception ex) {
-					LOGGER.error("error receiving response {} for {}", response, exchange, ex);
+				} catch (RuntimeException ex) {
+					LOGGER.warn("error receiving response {} for {}", response, exchange, ex);
 				}
 				reject(response, receiver);
 			}
@@ -433,8 +443,8 @@ public void run() {
 								"ignoring potentially forged reply for message {} with non-matching endpoint context",
 								idByMID);
 					}
-				} catch (Exception ex) {
-					LOGGER.error("error receiving empty message {} for {}", message, exchange, ex);
+				} catch (RuntimeException ex) {
+					LOGGER.warn("error receiving empty message {} for {}", message, exchange, ex);
 				}
 			}
 		});
@@ -444,11 +454,11 @@ private void reject(final Response response, final EndpointReceiver receiver) {
 
 		if (response.getType() != Type.ACK && response.hasMID()) {
 			// reject only messages with MID, ignore for TCP
-			LOGGER.debug("rejecting unmatchable response from {}", response.getSourceContext());
+			LOGGER.debug("rejecting response from {}", response.getSourceContext());
 			receiver.reject(response);
 		}
 	}
-	
+
 	private class RemoveHandlerImpl implements RemoveHandler {
 
 		@Override

californium-core/src/main/java/org/eclipse/californium/core/network/stack/BaseCoapStack.java

@@ -81,19 +81,39 @@ protected final void setLayers(final Layer specificLayers[]) {
 	@Override
 	public void sendRequest(final Exchange exchange, final Request request) {
 		// delegate to top
-		top.sendRequest(exchange, request);
+		try {
+			top.sendRequest(exchange, request);
+		} catch (RuntimeException ex) {
+			request.setSendError(ex);
+		}
 	}
 
 	@Override
 	public void sendResponse(final Exchange exchange, final Response response) {
 		// delegate to top
-		top.sendResponse(exchange, response);
+		try {
+			if (exchange.getRequest().getOptions().hasObserve()) {
+				// observe- or cancel-observe-requests may have
+				// multiple responses.
+				// when observes are finished, the last response has
+				// no longer an observe option. Therefore check the
+				// request for it.
+				exchange.retransmitResponse();
+			}
+			top.sendResponse(exchange, response);
+		} catch (RuntimeException ex) {
+			response.setSendError(ex);
+		}
 	}
 
 	@Override
 	public void sendEmptyMessage(final Exchange exchange, final EmptyMessage message) {
 		// delegate to top
-		top.sendEmptyMessage(exchange, message);
+		try {
+			top.sendEmptyMessage(exchange, message);
+		} catch (RuntimeException ex) {
+			message.setSendError(ex);
+		}
 	}
 
 	@Override

californium-core/src/main/java/org/eclipse/californium/core/network/stack/Block1BlockwiseStatus.java

@@ -120,9 +120,11 @@ public synchronized Request getNextRequestBlock() {
 		block.setDestinationContext(request.getDestinationContext());
 		// copy options
 		block.setOptions(new OptionSet(request.getOptions()));
-		// copy message observers so that a failing blockwise request also notifies observers registered with
-		// the original request
+		// copy message observers so that a failing blockwise request
+		// also notifies observers registered with the original request
 		block.addMessageObservers(request.getMessageObservers());
+		block.addMessageObserver(new FailureForwardingMessageObserver(request));
+
 		if (num == 0) {
 			// indicate overall body size to peer
 			block.getOptions().setSize1(request.getPayloadSize());

californium-core/src/main/java/org/eclipse/californium/core/network/stack/BlockwiseLayer.java

@@ -808,15 +808,25 @@ private void sendNextBlock(final Exchange exchange, final Response response, fin
 		}
 		int nextNum = status.getCurrentNum() + currentSize / newSize;
 		LOGGER.debug("sending next Block1 num={}", nextNum);
-		Request nextBlock = status.getNextRequestBlock(nextNum, newSzx);
-		// we use the same token to ease traceability
-		nextBlock.setToken(response.getToken());
-		nextBlock.setDestinationContext(response.getSourceContext());
-		addBlock1CleanUpObserver(nextBlock, key, status);
-
-		exchange.setCurrentRequest(nextBlock);
-		prepareBlock1Cleanup(status, key);
-		lower().sendRequest(exchange, nextBlock);
+		Request nextBlock = null;
+		try {
+			nextBlock = status.getNextRequestBlock(nextNum, newSzx);
+			// we use the same token to ease traceability
+			nextBlock.setToken(response.getToken());
+			nextBlock.setDestinationContext(response.getSourceContext());
+			addBlock1CleanUpObserver(nextBlock, key, status);
+
+			exchange.setCurrentRequest(nextBlock);
+			prepareBlock1Cleanup(status, key);
+			lower().sendRequest(exchange, nextBlock);
+		} catch (RuntimeException ex) {
+			LOGGER.warn("cannot process next block request, aborting request!", ex);
+			if (nextBlock != null) {
+				nextBlock.setSendError(ex);
+			} else {
+				exchange.getRequest().setSendError(ex);
+			}
+		}
 	}
 
 	/**
@@ -894,45 +904,51 @@ private void handleBlock2Response(final Exchange exchange, final Response respon
 					Request request = exchange.getRequest();
 
 					Request block = new Request(request.getCode());
-					// do not enforce CON, since NON could make sense over SMS or similar transports
-					block.setType(request.getType());
-					block.setDestinationContext(response.getSourceContext());
-
-					/*
-					 * WARNING:
-					 * 
-					 * For Observe, the Matcher then will store the same
-					 * exchange under a different KeyToken in exchangesByToken,
-					 * which is cleaned up in the else case below.
-					 */
-					if (!response.isNotification()) {
-						block.setToken(response.getToken());
-					} else if (exchange.isNotification()) {
-						// Recreate cleanup message observer 
-						request.addMessageObserver(new CleanupMessageObserver(exchange));
-					}
-
-					// copy options
-					block.setOptions(new OptionSet(request.getOptions()));
-					block.getOptions().setBlock2(newSzx, false, nextNum);
-
-					// make sure NOT to use Observe for block retrieval
-					block.getOptions().removeObserve();
-
-					// copy message observers from original request so that they will be notified
-					// if something goes wrong with this blockwise request, e.g. if it times out
-					block.addMessageObservers(request.getMessageObservers());
-					// add an observer that cleans up the block2 transfer tracker if the
-					// block request fails
-					addBlock2CleanUpObserver(block, key, status);
-
-
-					status.setCurrentNum(nextNum);
+					block.addMessageObserver(new FailureForwardingMessageObserver(request));
+					try {
+
+						// do not enforce CON, since NON could make sense over SMS or similar transports
+						block.setType(request.getType());
+						block.setDestinationContext(response.getSourceContext());
+
+						/*
+						 * WARNING:
+						 * 
+						 * For Observe, the Matcher then will store the same
+						 * exchange under a different KeyToken in exchangesByToken,
+						 * which is cleaned up in the else case below.
+						 */
+						if (!response.isNotification()) {
+							block.setToken(response.getToken());
+						} else if (exchange.isNotification()) {
+							// Recreate cleanup message observer 
+							request.addMessageObserver(new CleanupMessageObserver(exchange));
+						}
 
-					LOGGER.debug("requesting next Block2 [num={}]: {}", nextNum, block);
-					exchange.setCurrentRequest(block);
-					prepareBlock2Cleanup(status, key);
-					lower().sendRequest(exchange, block);
+						// copy options
+						block.setOptions(new OptionSet(request.getOptions()));
+						block.getOptions().setBlock2(newSzx, false, nextNum);
+
+						// make sure NOT to use Observe for block retrieval
+						block.getOptions().removeObserve();
+
+						// copy message observers from original request so that they will be notified
+						// if something goes wrong with this blockwise request, e.g. if it times out
+						block.addMessageObservers(request.getMessageObservers());
+						// add an observer that cleans up the block2 transfer tracker if the
+						// block request fails
+						addBlock2CleanUpObserver(block, key, status);
+
+						status.setCurrentNum(nextNum);
+
+						LOGGER.debug("requesting next Block2 [num={}]: {}", nextNum, block);
+						exchange.setCurrentRequest(block);
+						prepareBlock2Cleanup(status, key);
+						lower().sendRequest(exchange, block);
+					} catch (RuntimeException ex) {
+						LOGGER.warn("cannot process next block request, aborting request!", ex);
+						block.setSendError(ex);
+					}
 
 				} else {
 

californium-core/src/main/java/org/eclipse/californium/core/network/stack/FailureForwardingMessageObserver.java

@@ -0,0 +1,59 @@
+/*******************************************************************************
+ * Copyright (c) 2019 Bosch Software Innovations GmbH and others.
+ * 
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ * 
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ *    http://www.eclipse.org/org/documents/edl-v10.html.
+ * 
+ * Contributors:
+ *    Bosch Software Innovations GmbH - initial creation
+ ******************************************************************************/
+package org.eclipse.californium.core.network.stack;
+
+import org.eclipse.californium.core.coap.Message;
+import org.eclipse.californium.core.coap.MessageObserverAdapter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Forward failures to other messages.
+ * 
+ * Forward {@link #onCancel()}, {@link #onReject()}, {@link #onTimeout()}, and
+ * {@link #onSendError(Throwable)} to provided message. Intended to be used for
+ * blockwise transfers
+ */
+public class FailureForwardingMessageObserver extends MessageObserverAdapter {
+
+	protected static final Logger LOGGER = LoggerFactory.getLogger(FailureForwardingMessageObserver.class.getName());
+
+	protected final Message message;
+
+	protected FailureForwardingMessageObserver(final Message message) {
+		this.message = message;
+	}
+
+	@Override
+	public void onCancel() {
+		message.cancel();
+	}
+
+	@Override
+	public void onReject() {
+		message.setRejected(true);
+	}
+
+	@Override
+	public void onTimeout() {
+		message.setTimedOut(true);
+	}
+
+	@Override
+	public void onSendError(Throwable error) {
+		message.setSendError(error);
+	}
+}