使用角度的一个客户端上的多个套接字

Question

我什至不确定这是否可能。如果没有，请指出我不同的方向。

我有一个允许多个 socket-io 连接的服务器。服务器生成一个随机数并发送到注册的套接字。

在客户端，我想初始化两个套接字并显示从服务器生成的相同数据。这两个套接字显示在一个模板上。

服务器

const express = require('express'),
      app = express(),
      server = require('http').createServer(app);
      io = require('socket.io')(server);

let timerId = null,
    sockets = new Set();

//This example emits to individual sockets (track by sockets Set above).
//Could also add sockets to a "room" as well using socket.join('roomId')
//https://socket.io/docs/server-api/#socket-join-room-callback

app.use(express.static(__dirname + '/dist')); 

/*
everytime there is a new connection, there is a new socket
*/
io.on('connection', socket => {

  sockets.add(socket);
  console.log(`Socket ${socket.id} added`);

  if (!timerId) {
    startTimer();
  }

  socket.on('clientdata', data => {
    console.log(data);
  });

  socket.on('disconnect', () => {
    console.log(`Deleting socket: ${socket.id}`);
    sockets.delete(socket);
    console.log(`Remaining sockets: ${sockets.size}`);
  });

});

/**
 * for each socket, we emit the same value
 */
function startTimer() {
  //Simulate stock data received by the server that needs 
  //to be pushed to clients
  timerId = setInterval(() => {
    if (!sockets.size) {
      clearInterval(timerId);
      timerId = null;
      console.log(`Timer stopped`);
    }
    let value = ((Math.random() * 50) + 1).toFixed(2);
    //See comment above about using a "room" to emit to an entire
    //group of sockets if appropriate for your scenario
    //This example tracks each socket and emits to each one
    for (const s of sockets) {
      console.log(`Emitting value: ${value}`);
      s.emit('data', { data: value });
    }

  }, 2000);
}

server.listen(8080);
console.log('Visit http://localhost:8080 in your browser');

客户

app.component.ts

这是我试图解决这个问题的地方。对于两个不同的套接字，我有两个不同的 stockQuote 变量和两个不同的订阅。

import { Component, OnInit, OnDestroy } from '@angular/core';
import { DataService } from './core/data.service';
import { Subscription } from 'rxjs';

@Component({
  selector: 'app-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.css']
})
export class AppComponent {
  title = 'health-status-dashboard';

  stockQuote: number;
  stockQuote1: number;
  sub: Subscription;
  sub1: Subscription;

  constructor(private dataService: DataService) { }

  ngOnInit() {
    this.sub = this.dataService.getQuotes()
      .subscribe(quote => {
        this.stockQuote = quote;
      });

    this.sub1 = this.dataService.getQuotes()
      .subscribe(quote => {
        this.stockQuote1 = quote;
      });
  }

  ngOnDestroy() {
    this.sub.unsubscribe();
  }
}

app.component.html

<h2>Quote: {{ stockQuote }}</h2>
<h2>Quote1: {{ stockQuote1 }}</h2>

服务

在这里，我尝试在每次调用 getQuotes() 时创建一个新套接字。 （在 app.component.ts 的 ngOnInit() 方法中调用了两次。因此创建了两个套接字。

import { Injectable } from '@angular/core';
//TODO: why cannot import?
import { Observable } from 'rxjs';
import { Observer } from 'rxjs';
import { map, catchError } from 'rxjs/operators';
import * as socketIo from 'socket.io-client';

import { Socket } from '../shared/interfaces';

declare var io: {
  connect(url: string): Socket;
}

/**
 * TODO: need to test
 */
function isError(error: Error): error is Error {
  return (<Error>error) !== undefined;
}

@Injectable()
export class DataService {
  sockets: Socket[] = [];
  observer: Observer<number>;

  /**
   * previously: there was one socket variable and a new socket was created each time still but just overwrote
   the previous socket. still, the previous socket exists.
   */
  getQuotes(): Observable<number> {
    const soc: Socket = socketIo('http://localhost:8080');
    //this.sockets.push(soc);

    soc.on('data', (res) => {
      this.observer.next(res.data);
    });

    return this.createObservable();
  }

  createObservable(): Observable<number> {
    return new Observable(observer => {
      this.observer = observer;
    });
  }

  private handleError(error) {
    console.error('server error:', error);
    if (isError(error.error)) {
      let errMessage = error.error.message;
      return Observable.throw(errMessage);
    }
    return Observable.throw(error || 'Socket.io server error');
  }
}

interfaces.ts

export interface Socket {
  on(event: string, callback: (data: any) => void);
  emit(event: string, data: any);
}

在我的应用中，只显示stockQuote1（模板中的Quote1），stockQuote 为空白（模板中的Quote）。

我希望 Quote 和 Quote1 在模板中显示相同的值。

在我的日志中说添加了两个套接字：

...
Socket s_-VWv34H6Vd7nN1AAAA added
Socket lENrl8vr_YvplUygAAAB added
...

Answer 1

@Injectable()
export class DataService {

  /**
   * previously: there was one socket variable and a new socket was created each time still but just overwrote
   the previous socket. still, the previous socket exists.
   */
  getQuotes(): Observable<number> {
    const soc = socketIo('http://localhost:8080');
    //this.sockets.push(soc);
    //create new Observable in function, and use `observer`
    return new Observable(observer => {
      soc.on('data', (res) => {
        observer.next(res.data);
      });
    });
  }
}

原因：
套接字可以连接多个。
但是，observer 是一，所以只有最后一个套接字的值会到达。
我的代码在getQuotes() 方法中使用本地observer。

对于类变量，我们需要Observable<number>的数组