Use removeEach to clean up unsettled map

Author: acogoluegnes

src/main/qpid/org/apache/qpid/protonj2/engine/impl/ProtonReceiver.java

@@ -21,6 +21,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
+import java.util.function.Consumer;
 import java.util.function.Predicate;
 
 import org.apache.qpid.protonj2.buffer.ProtonBuffer;
@@ -46,6 +47,12 @@
  */
 public class ProtonReceiver extends ProtonLink<Receiver> implements Receiver {
 
+    private static final Consumer<ProtonIncomingDelivery> RELEASE_INCOMING_DELIVERY_TAG_CALLBACK = d -> {
+        if (d.getTag() != null) {
+            d.getTag().release();
+        }
+    };
+
     private EventHandler<IncomingDelivery> deliveryReadEventHandler = null;
     private EventHandler<IncomingDelivery> deliveryAbortedEventHandler = null;
     private EventHandler<IncomingDelivery> deliveryUpdatedEventHandler = null;
@@ -243,10 +250,7 @@ public void disposition(DeliveryState state, long[] range) {
         try {
             sessionWindow.processDisposition(state, range);
         } finally {
-            for (long i = range[0]; i < range[1]; i++) {
-                unsettled.remove((int) i);
-            }
-            // TODO release delivery tags
+            unsettled.removeEach((int) range[0], (int) range[1], RELEASE_INCOMING_DELIVERY_TAG_CALLBACK);
         }
     }
 

src/main/qpid/org/apache/qpid/protonj2/engine/impl/ProtonSessionIncomingWindow.java

@@ -232,9 +232,7 @@ void processDisposition(ProtonReceiver receiver, ProtonIncomingDelivery delivery
     }
 
     void processDisposition(DeliveryState state, long [] range) {
-        for (long i = range[0]; i < range[1]; i++) {
-            unsettled.remove((int) i);
-        }
+        unsettled.removeEach((int) range[0], (int) range[1], d -> { });
         cachedDisposition.reset();
         cachedDisposition.setFirst(range[0]);
         cachedDisposition.setLast(range[1]);