ReFlex Client: Stream sensor data with SignalR

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
import { Injectable, OnDestroy } from '@angular/core';
import {
  HubConnection, HubConnectionBuilder, HubConnectionState, LogLevel
} from '@microsoft/signalr';
import { BehaviorSubject, Subject } from 'rxjs';

export interface Point3 {
  x: number;
  y: number;
  z: number;
  isValid: boolean;
  isFiltered: boolean;
}

@Injectable()
export class SensorStreamService implements OnDestroy {
  private connection?: HubConnection;
  private destroyed = false;
  private readonly state = new BehaviorSubject(HubConnectionState.Disconnected);
  private readonly active = new BehaviorSubject(false);
  private readonly error = new BehaviorSubject<string | null>(null);
  private readonly frames = new Subject<Point3[]>();

  readonly state$ = this.state.asObservable();
  readonly active$ = this.active.asObservable();
  readonly error$ = this.error.asObservable();
  readonly points$ = this.frames.asObservable();

  async connect(hubUrl: string): Promise<void> {
    if (this.destroyed || this.connection) return;
    const connection = new HubConnectionBuilder()
      .withUrl(hubUrl)
      .withAutomaticReconnect([0, 2000, 10000, 30000])
      .configureLogging(LogLevel.Warning)
      .build();
    this.connection = connection;
    this.error.next(null);
    this.state.next(HubConnectionState.Connecting);

    // Ignore callbacks from a connection that has already been released.
    const current = () => this.connection === connection;
    connection.on('pointCloud', (points: Point3[]) => {
      if (current()) this.frames.next(points);
    });
    connection.onreconnecting(error => {
      if (!current()) return;
      this.active.next(false);
      this.state.next(HubConnectionState.Reconnecting);
      this.report(error ?? 'Connection lost; reconnecting.');
    });
    connection.onreconnected(() => {
      if (!current()) return;
      // The server sees a new connection: subscribe again.
      void this.activate(connection);
    });
    connection.onclose(error => {
      if (!current()) return;
      this.connection = undefined;
      connection.off('pointCloud');
      this.active.next(false);
      this.state.next(HubConnectionState.Disconnected);
      this.report(error ?? 'Connection closed. Select Connect to retry.');
    });

    try {
      await connection.start();
      if (current()) await this.activate(connection);
    } catch (error) {
      if (!current()) return;
      this.report(error);
      await this.disconnect();
    }
  }

  private async activate(connection: HubConnection): Promise<void> {
    this.state.next(connection.state);
    try {
      await connection.invoke('startPointCloud');
      if (this.connection !== connection) return;
      this.active.next(true);
      this.error.next(null);
    } catch (error) {
      if (this.connection !== connection) return;
      this.report(error);
      // A failed subscription must not look like an active stream.
      await this.disconnect();
    }
  }

  async disconnect(): Promise<void> {
    const connection = this.connection;
    if (!connection) return;
    this.connection = undefined;
    this.active.next(false);
    this.state.next(HubConnectionState.Disconnecting);
    connection.off('pointCloud');
    try {
      if (connection.state === HubConnectionState.Connected) {
        await connection.invoke('stopPointCloud');
      }
    } catch (error) {
      this.report(error);
    } finally {
      // Always stop, including during connection/reconnection or RPC failure.
      try {
        await connection.stop();
      } catch (error) {
        this.report(error);
      }
      if (!this.connection) this.state.next(HubConnectionState.Disconnected);
    }
  }

  private report(error: unknown): void {
    this.error.next(error instanceof Error ? error.message : String(error));
  }

  ngOnDestroy(): void {
    this.destroyed = true;
    void this.disconnect().finally(() => {
      this.frames.complete();
      this.state.complete();
      this.active.complete();
      this.error.complete();
    });
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
import { CommonModule } from '@angular/common';
import { Component, OnDestroy } from '@angular/core';
import { SensorStreamService } from './sensor-stream.service';

@Component({
  selector: 'app-sensor-viewer',
  standalone: true,
  imports: [CommonModule],
  providers: [SensorStreamService],
  template: `
    <p>Hub: {{ stream.state$ | async }}</p>
    <p>Subscription active: {{ stream.active$ | async }}</p>
    <p *ngIf="stream.error$ | async as error" role="alert">{{ error }}</p>
    <button (click)="connect()"
      [disabled]="(stream.state$ | async) !== 'Disconnected'">Connect</button>
    <button (click)="disconnect()"
      [disabled]="(stream.state$ | async) === 'Disconnected' ||
                  (stream.state$ | async) === 'Disconnecting'">Disconnect</button>
    <p>Points in latest frame: {{ pointCount }}</p>
  `
})
export class SensorViewerComponent implements OnDestroy {
  pointCount = 0;
  private readonly frames;

  constructor(readonly stream: SensorStreamService) {
    // Subscribe before Connect can activate delivery.
    this.frames = stream.points$.subscribe(points => {
      this.pointCount = points.length;
      // Forward points to the renderer here (Step 4).
    });
  }

  connect(): void {
    void this.stream.connect('/pointcloudhub');
  }

  disconnect(): void {
    this.pointCount = 0;
    void this.stream.disconnect();
  }

  ngOnDestroy(): void {
    this.frames.unsubscribe();
    // Angular also destroys the component-scoped service, which disconnects.
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
function updateCloud(cloud: THREE.Points, points: Point3[]): void {
  const visible = points.filter(p => p.isValid && !p.isFiltered &&
    Number.isFinite(p.x) && Number.isFinite(p.y) && Number.isFinite(p.z));
  const geometry = cloud.geometry;
  let position = geometry.getAttribute('position') as THREE.BufferAttribute;
  if (!position || position.count !== visible.length) {
    position = new THREE.BufferAttribute(new Float32Array(visible.length * 3), 3);
    geometry.setAttribute('position', position);
  }
  visible.forEach((p, i) => position.setXYZ(i, p.x * 10, p.y * -10, p.z * 10));
  position.needsUpdate = true;
  geometry.setDrawRange(0, visible.length);
  cloud.visible = visible.length > 0;
  if (cloud.visible) geometry.computeBoundingSphere();
}