Adds the concept of DataSource to Amphion

package.json

@@ -28,7 +28,8 @@
     "stats-js": "^1.0.1",
     "three-freeform-controls": "0.0.18",
     "three-spritetext": "^1.1.1",
-    "urdf-js": "0.0.7"
+    "urdf-js": "^0.0.14",
+    "xstream": "^11.11.0"
   },
   "peerDependencies": {
     "three": "^0.109.0"

src/core/index.js

@@ -1,4 +1,4 @@
-import ROSLIB from 'roslib';
+import { RosTopicDataSource } from '../data/ros';
 
 class Core {
   constructor(ros, topicName, messageType, options = {}) {
@@ -32,20 +32,25 @@ class Core {
   reset() {}
 
   subscribe() {
-    if (!this.topicInstances) {
+    if (!this.dataSourceInstances) {
       return;
     }
-    this.topicInstances.forEach(t => {
-      t.subscribe(this.update);
+    this.dataSourceInstances.forEach(t => {
+      const listener = {
+        next: this.update,
+        error: error => console.log(error),
+        complete: () => console.log('stream complete'),
+      };
+      t.addListener(listener);
     });
   }
 
   unsubscribe() {
-    if (!this.topicInstances) {
+    if (!this.dataSourceInstances) {
       return;
     }
-    this.topicInstances.forEach(t => {
-      t.unsubscribe();
+    this.dataSourceInstances.forEach(t => {
+      t.removeAllListeners();
     });
   }
 
@@ -73,18 +78,18 @@ class Core {
 
     this.topicName = topicName;
     this.messageType = type || this.messageType;
-    this.topicInstances = (Array.isArray(topicName)
+    this.dataSourceInstances = (Array.isArray(topicName)
       ? topicName
       : [{ name: topicName, messageType: type }]
     ).map(
       ({ name, messageType }) =>
-        new ROSLIB.Topic({
+        new RosTopicDataSource({
           ros: this.ros,
-          name,
+          topicName: name,
           messageType,
           compression: compression || 'none',
-          throttle_rate: throttleRate || 0,
-          queue_size: queueSize || 10,
+          throttleRate: throttleRate || 0,
+          queueSize: queueSize || 10,
         }),
     );
 

src/core/live.ts

@@ -0,0 +1,95 @@
+import { DataSource } from '../data';
+import { Object3D } from 'three';
+import { assertIsDefined, isHTMLElement, isObject3D } from '../utils/helpers';
+import { Listener } from 'xstream';
+
+interface CoreOptions<T> {
+  sources: Array<DataSource<T>>;
+  options?: { [k: string]: any };
+}
+
+class LiveCore<T extends RosMessage.Base, V extends Object3D | HTMLElement> {
+  private onHeaderChange = (headerFrameId: string) => {};
+  private sources: Array<DataSource<T>>;
+  protected options: { [k: string]: any };
+  private headerFrameId = '';
+  public object?: V;
+
+  constructor(args: CoreOptions<T>) {
+    this.sources = args.sources;
+    this.options = args.options ?? {};
+
+    this.update = this.update.bind(this);
+    this.updateOptions = this.updateOptions.bind(this);
+  }
+
+  hide = () => {
+    assertIsDefined(this.object);
+    if (isObject3D(this.object)) {
+      this.object.visible = false;
+    } else if (isHTMLElement(this.object)) {
+      this.object.style.visibility = 'hidden';
+    }
+  };
+
+  show = () => {
+    assertIsDefined(this.object);
+    if (isObject3D(this.object)) {
+      this.object.visible = true;
+    } else if (isHTMLElement(this.object)) {
+      this.object.style.visibility = 'visible';
+    }
+  };
+
+  destroy = () => {
+    this.unsubscribe();
+    if (isObject3D(this.object)) {
+      this.object?.parent?.remove(this.object);
+    } else if (isHTMLElement(this.object)) {
+      this.object?.parentElement?.removeChild(this.object);
+    }
+    this.object = undefined;
+  };
+
+  reset = () => {};
+
+  subscribe = () => {
+    this.sources.forEach(source => {
+      const listener: Listener<T> = {
+        next: this.update,
+        error: error => console.log(error),
+        complete: () => {},
+      };
+      source.addListener(listener);
+    });
+  };
+
+  unsubscribe = () => {
+    this.sources.forEach(source => {
+      source.removeAllListeners();
+    });
+  };
+
+  update(message: T) {
+    const headerFrameId = message.header?.frame_id ?? '';
+    if (headerFrameId !== this.headerFrameId) {
+      this.headerFrameId = headerFrameId;
+      this.onHeaderChange(this.headerFrameId);
+    }
+  }
+
+  updateOptions(options: { [k: string]: any }) {
+    this.options = {
+      ...this.options,
+      ...options,
+    };
+  }
+
+  changeSources(sources: Array<DataSource<T>>) {
+    this.unsubscribe();
+    this.sources = sources;
+    this.subscribe();
+  }
+}
+
+export default LiveCore;