Datagram drop callback

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

@@ -90,6 +90,12 @@ public boolean isOverflown() {
 		return overflow;
 	}
 
+	/**
+	 * Handles {@code RejectedExecutionException}
+	 * @param ex the thrown exception
+	 */
+	public void onError(RejectedExecutionException ex) {};
+
 	/**
 	 * Execute this job.
 	 * 
@@ -101,6 +107,7 @@ public void execute(Executor executor) {
 			executor.execute(this);
 		} catch (RejectedExecutionException ex) {
 			onDequeueing();
+			onError(ex);
 			throw ex;
 		}
 	}

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

@@ -1822,6 +1822,9 @@ protected void processDatagram(DatagramPacket packet, InetSocketAddress router)
 			// other information. If not used, a value of zero is inserted.
 			DROP_LOGGER.trace("Discarding record with {} bytes from [{}] without source-port", packet.getLength(),
 					StringUtil.toLog(peerAddress));
+			if (datagramFilter instanceof DatagramFilterExtended) {
+				((DatagramFilterExtended) datagramFilter).onDrop(packet);
+			}
 			if (health != null) {
 				health.receivingRecord(true);
 			}
@@ -1831,6 +1834,9 @@ protected void processDatagram(DatagramPacket packet, InetSocketAddress router)
 			if (!datagramFilter.onReceiving(packet)) {
 				DROP_LOGGER.trace("Filter out packet with {} bytes from [{}]", packet.getLength(),
 						StringUtil.toLog(peerAddress));
+				if (datagramFilter instanceof DatagramFilterExtended) {
+					((DatagramFilterExtended) datagramFilter).onDrop(packet);
+				}
 				if (health != null) {
 					health.receivingRecord(true);
 				}
@@ -1845,6 +1851,9 @@ protected void processDatagram(DatagramPacket packet, InetSocketAddress router)
 		if (records.isEmpty()) {
 			DROP_LOGGER.trace("Discarding malicious record with {} bytes from [{}]", packet.getLength(),
 					StringUtil.toLog(peerAddress));
+			if (datagramFilter instanceof DatagramFilterExtended) {
+				((DatagramFilterExtended) datagramFilter).onDrop(packet);
+			}
 			if (health != null) {
 				health.receivingRecord(true);
 			}
@@ -1856,6 +1865,9 @@ protected void processDatagram(DatagramPacket packet, InetSocketAddress router)
 					records.get(0).getType(), StringUtil.toLog(peerAddress));
 			LOGGER.debug("Execution shutdown while processing incoming records from peer: {}",
 					StringUtil.toLog(peerAddress));
+			if (datagramFilter instanceof DatagramFilterExtended) {
+				((DatagramFilterExtended) datagramFilter).onDrop(packet);
+			}
 			if (health != null) {
 				health.receivingRecord(true);
 			}
@@ -1885,6 +1897,9 @@ protected void processRecords(final List<Record> records,
 			if (dtlsRole == DtlsRole.CLIENT_ONLY) {
 				DROP_LOGGER.trace("client-only, discarding {} CLIENT_HELLO from [{}]!", records.size(),
 						StringUtil.toLog(peerAddress));
+				if (datagramFilter instanceof DatagramFilterExtended) {
+					((DatagramFilterExtended) datagramFilter).onDrop(firstRecord);
+				}
 				if (health != null) {
 					health.receivingRecord(true);
 				}
@@ -1908,6 +1923,13 @@ public void run() {
 						onDequeueing();
 					}
 				}
+
+				@Override
+				public void onError(RejectedExecutionException ex) {
+					if (datagramFilter instanceof DatagramFilterExtended) {
+						((DatagramFilterExtended) datagramFilter).onDrop(firstRecord);
+					}
+				}
 			});
 			return;
 		}
@@ -1916,6 +1938,9 @@ public void run() {
 		final Connection connection = getConnection(peerAddress, connectionId, false);
 
 		if (connection == null) {
+			if (datagramFilter instanceof DatagramFilterExtended) {
+				((DatagramFilterExtended) datagramFilter).onDrop(firstRecord);
+			}
 			if (health != null) {
 				health.receivingRecord(true);
 			}
@@ -1935,7 +1960,6 @@ public void run() {
 			record.setAddress(peerAddress, router);
 			try {
 				if (!executeInbound(serialExecutor, peerAddress, new LimitedRunnable(pendingInboundJobsCountdown) {
-
 					@Override
 					public void run() {
 						try {
@@ -1946,6 +1970,13 @@ public void run() {
 							onDequeueing();
 						}
 					}
+
+					@Override
+					public void onError(RejectedExecutionException ex) {
+						if (datagramFilter instanceof DatagramFilterExtended) {
+							((DatagramFilterExtended) datagramFilter).onDrop(record);
+						}
+					}
 				})) {
 					break;
 				}
@@ -2005,6 +2036,9 @@ public void processRecord(Record record, Connection connection) {
 				DROP_LOGGER.debug("Drop received record {}, connection changed address {} => {}! (shift {}ms)",
 						record.getType(), StringUtil.toLog(record.getPeerAddress()),
 						StringUtil.toLog(connection.getPeerAddress()), delay);
+				if (datagramFilter instanceof DatagramFilterExtended) {
+					((DatagramFilterExtended) datagramFilter).onDrop(record);
+				}
 				if (health != null) {
 					health.receivingRecord(true);
 				}
@@ -2027,6 +2061,9 @@ public void processRecord(Record record, Connection connection) {
 							"Discarding {} record [epoch {}, rseqn {}] received from peer [{}], handshake expired!",
 							record.getType(), epoch, record.getSequenceNumber(),
 							StringUtil.toLog(record.getPeerAddress()));
+					if (datagramFilter instanceof DatagramFilterExtended) {
+						((DatagramFilterExtended) datagramFilter).onDrop(record);
+					}
 					if (health != null) {
 						health.receivingRecord(true);
 					}
@@ -2046,6 +2083,9 @@ public void processRecord(Record record, Connection connection) {
 							"Discarding {} record [epoch {}, rseqn {}] received from peer [{}] without an active dtls context",
 							record.getType(), epoch, record.getSequenceNumber(),
 							StringUtil.toLog(record.getPeerAddress()));
+					if (datagramFilter instanceof DatagramFilterExtended) {
+						((DatagramFilterExtended) datagramFilter).onDrop(record);
+					}
 					if (health != null) {
 						health.receivingRecord(true);
 					}
@@ -2071,6 +2111,9 @@ public void processRecord(Record record, Connection connection) {
 							record.getType(), epoch, record.getSequenceNumber(),
 							StringUtil.toLog(record.getPeerAddress()));
 				}
+				if (datagramFilter instanceof DatagramFilterExtended) {
+					((DatagramFilterExtended) datagramFilter).onDrop(record);
+				}
 				if (health != null) {
 					health.receivingRecord(true);
 				}
@@ -2082,6 +2125,9 @@ public void processRecord(Record record, Connection connection) {
 				if (epoch == 0) {
 					DROP_LOGGER.debug("Discarding TLS_CID record received from peer [{}] during handshake",
 							StringUtil.toLog(record.getPeerAddress()));
+					if (datagramFilter instanceof DatagramFilterExtended) {
+						((DatagramFilterExtended) datagramFilter).onDrop(record);
+					}
 					if (health != null) {
 						health.receivingRecord(true);
 					}
@@ -2090,6 +2136,9 @@ public void processRecord(Record record, Connection connection) {
 			} else if (epoch > 0 && connection.expectCid()) {
 				DROP_LOGGER.debug("Discarding record received from peer [{}], CID required!",
 						StringUtil.toLog(record.getPeerAddress()));
+				if (datagramFilter instanceof DatagramFilterExtended) {
+					((DatagramFilterExtended) datagramFilter).onDrop(record);
+				}
 				if (health != null) {
 					health.receivingRecord(true);
 				}
@@ -2101,6 +2150,9 @@ public void processRecord(Record record, Connection connection) {
 					if (!datagramFilter.onReceiving(record, connection)) {
 						DROP_LOGGER.trace("Filter out record with {} bytes from [{}]", record.size(),
 								StringUtil.toLog(record.getPeerAddress()));
+						if (datagramFilter instanceof DatagramFilterExtended) {
+							((DatagramFilterExtended) datagramFilter).onDrop(record);
+						}
 						if (health != null) {
 							health.receivingRecord(true);
 						}
@@ -2140,6 +2192,9 @@ public void processRecord(Record record, Connection connection) {
 						StringUtil.toLog(record.getPeerAddress()));
 			}
 		} catch (RuntimeException e) {
+			if (datagramFilter instanceof DatagramFilterExtended) {
+				((DatagramFilterExtended) datagramFilter).onDrop(record);
+			}
 			if (health != null) {
 				health.receivingRecord(true);
 			}
@@ -2163,6 +2218,9 @@ public void processRecord(Record record, Connection connection) {
 			}
 			DROP_LOGGER.debug("Discarding {} received from peer [{}] caused by {}", record.getType(),
 					StringUtil.toLog(record.getPeerAddress()), e.getMessage());
+			if (datagramFilter instanceof DatagramFilterExtended) {
+				((DatagramFilterExtended) datagramFilter).onDrop(record);
+			}
 			if (health != null) {
 				if (health instanceof DtlsHealthExtended2) {
 					((DtlsHealthExtended2) health).receivingMacError();
@@ -2173,6 +2231,9 @@ public void processRecord(Record record, Connection connection) {
 		} catch (GeneralSecurityException e) {
 			DROP_LOGGER.debug("Discarding {} received from peer [{}] caused by {}", record.getType(),
 					StringUtil.toLog(record.getPeerAddress()), e.getMessage());
+			if (datagramFilter instanceof DatagramFilterExtended) {
+				((DatagramFilterExtended) datagramFilter).onDrop(record);
+			}
 			if (health != null) {
 				health.receivingRecord(true);
 			}
@@ -2230,6 +2291,9 @@ private void processApplicationDataRecord(final Record record, final Connection
 			if (useNewerRecordFilter && !newest) {
 				DROP_LOGGER.debug("Discarding reorderd {} record [epoch {}, rseqn {}] received from peer [{}]",
 						record.getType(), record.getEpoch(), record.getSequenceNumber(), StringUtil.toLog(record.getPeerAddress()));
+				if (datagramFilter instanceof DatagramFilterExtended) {
+					((DatagramFilterExtended) datagramFilter).onDrop(record);
+				}
 				if (health != null) {
 					health.receivingRecord(true);
 				}

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

@@ -0,0 +1,39 @@
+/*******************************************************************************
+ * Copyright (c) 2022 AVSystem and others.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v2.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v20.html
+ * and the Eclipse Distribution License is available at
+ *    http://www.eclipse.org/org/documents/edl-v10.html.
+ *
+ * Contributors:
+ *    Bartłomiej Wiśniewski (AVSystem) - initial creation
+ ******************************************************************************/
+package org.eclipse.californium.scandium;
+
+import org.eclipse.californium.elements.util.PublicAPIExtension;
+import org.eclipse.californium.scandium.dtls.Record;
+
+import java.net.DatagramPacket;
+
+/**
+ * Extension of DatagramFilter
+ */
+@PublicAPIExtension(type = DatagramFilter.class)
+public interface DatagramFilterExtended {
+
+	/**
+	 * Called when a datagram packed is dropped. Allows to inject packet based action in form of callback
+	 * @param packet the dropped datagram packet
+	 */
+	void onDrop(DatagramPacket packet);
+	/**
+	 * Called when a record is dropped. Allows to inject record based action in form of callback
+	 * @param record the dropped record
+	 */
+	void onDrop(Record record);
+}