1 files changed, 28 insertions, 3 deletions
diff --git a/src/com/benburwell/planes/sbs/AggregateDataSource.java b/src/com/benburwell/planes/sbs/AggregateDataSource.java
index 08b89ea..2850404 100644
--- a/src/com/benburwell/planes/sbs/AggregateDataSource.java
+++ b/src/com/benburwell/planes/sbs/AggregateDataSource.java
@@ -2,22 +2,47 @@ package com.benburwell.planes.sbs;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
+import java.util.HashMap;
 
 /**
  * Created by ben on 11/15/16.
  */
 public class AggregateDataSource implements DataSource {
     private List<DataListener> subscribers = new ArrayList<>();
+    private boolean isOpen = false;
+    private int nextSourceNumber = 1;
+    private Map<Integer,DataSource> sources = new HashMap<>();
 
-    public void addSource(DataSource source) {
+    public int addSource(DataSource source) {
+        int thisSourceNumber = this.nextSourceNumber++;
+        this.sources.put(thisSourceNumber, source);
         source.subscribe((SBSPacket packet) -> {
-            for (DataListener listener : subscribers) {
-                listener.handleMessage(packet);
+            if (isOpen) {
+                for (DataListener listener : subscribers) {
+                    listener.handleMessage(packet);
+                }
             }
         });
+        return thisSourceNumber;
     }
 
     public void subscribe(DataListener listener) {
         this.subscribers.add(listener);
     }
+
+    public void open() {
+        this.isOpen = true;
+    }
+
+    public void close() {
+        this.isOpen = false;
+    }
+
+    public void closeSource(int sourceNumber) {
+        if (this.sources.containsKey(sourceNumber)) {
+            this.sources.get(sourceNumber).close();
+            this.sources.remove(sourceNumber);
+        }
+    }
 }