Public receive method in SBPFramer Java class

java/src/com/swiftnav/sbp/client/SBPCallback.java

@@ -16,4 +16,5 @@
 /** Interface for SBP message handlers. */
 public interface SBPCallback {
     void receiveCallback(SBPMessage msg);
+    void callbacksDone();
 }

java/src/com/swiftnav/sbp/client/SBPHandler.java

@@ -23,6 +23,8 @@
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /** Provide an interface for queueing and filtering messages.
  */
@@ -148,6 +150,17 @@ public void run() {
                         dispatch(null, msg);
                 }
             }
+
+            // Inform all callbacks that there are no more messages coming from the source
+            for (List<Reference<SBPCallback>> cb_list : callbacks.values()) {
+                for (Reference<SBPCallback> wr : cb_list) {
+                    SBPCallback cb = wr.get();
+                    if (cb != null) {
+                        cb.callbacksDone();
+                    }
+                }
+            }
+
         }
 
         void finish() {
@@ -169,27 +182,42 @@ private void dispatch(Integer type, SBPMessage msg) {
 
     class SBPQueueIterator extends SBPIterable implements SBPCallback {
         private BlockingQueue<SBPMessage> queue;
+        private AtomicBoolean finished;
 
         private SBPQueueIterator(int queue_size) {
             if (queue_size <= 0) {
                 queue = new LinkedBlockingQueue<>();
             } else {
                 queue = new ArrayBlockingQueue<>(queue_size);
             }
+            finished = new AtomicBoolean(false);
         }
 
         @Override
         public void receiveCallback(SBPMessage msg) {
             queue.add(msg);
         }
 
+        @Override
+        public void callbacksDone() {
+            finished.set(true);
+        }
+
         @Override
         protected SBPMessage getNext() {
-            try {
-                return queue.take();
-            } catch (InterruptedException e) {
-                throw new NoSuchElementException();
+            while (!finished.get()) {
+                try {
+                    SBPMessage msg = queue.poll(100, TimeUnit.MILLISECONDS);
+                    if (msg != null) {
+                        return msg;
+                    }
+                    continue;
+                } catch (InterruptedException e) {
+                    continue;
+                }
             }
+            // If we get here finished is set to true so there are no more messages available
+            throw new NoSuchElementException();
         }
     }
 }