我的角应用程序使用websocket与后端进行通信.
在我的测试用例中,我有2个客户端组件.Observable计时器按预期打印两个不同的客户端ID.
每个ngOnInit()还打印其客户端的id.
现在出于某种原因,对于每条消息,websocketService.observeClient()的订阅被调用2次,但this.client.id总是打印第二个客户端的值.
继承我的客户组件
@Component({
...
})
export class ClientComponent implements OnInit {
@Input() client: Client;
constructor(public websocketService: WebsocketService) {
Observable.timer(1000, 1000).subscribe(() => console.log(this.client.id));
}
ngOnInit() {
console.log(this.client.id);
this.websocketService.observeClient().subscribe(data => {
console.log('message', this.client.id);
});
}
}
和我的websocket服务
@Injectable()
export class WebsocketService {
private observable: Observable<MessageEvent>;
private observer: Subject<Message>;
constructor() {
const socket = new WebSocket('ws://localhost:9091');
this.observable = Observable.create(
(observer: Observer<MessageEvent>) => {
socket.onmessage = observer.next.bind(observer);
socket.onerror = observer.error.bind(observer);
socket.onclose = observer.complete.bind(observer);
return socket.close.bind(socket);
}
);
this.observer = …我将使用Angular2接收websocket传入消息并根据收到的消息更新网页.现在,我正在使用虚拟回声websocket服务并将替换它.
根据我的理解,接收websocket消息的函数必须返回由将更新网页的处理程序订阅的observable.但我无法弄清楚如何返回一个可观察的.
代码段附于下方.在MonitorService创建一个WebSocket连接,并返回可观察到包含接收到的消息.
@Injectable()
export class MonitorService {
private actionUrl: string;
private headers: Headers;
private websocket: any;
private receivedMsg: any;
constructor(private http: Http, private configuration: AppConfiguration) {
this.actionUrl = configuration.BaseUrl + 'monitor/';
this.headers = new Headers();
this.headers.append('Content-Type', 'application/json');
this.headers.append('Accept', 'application/json');
}
public GetInstanceStatus = (): Observable<Response> => {
this.websocket = new WebSocket("ws://echo.websocket.org/"); //dummy echo websocket service
this.websocket.onopen = (evt) => {
this.websocket.send("Hello World");
};
this.websocket.onmessage = (evt) => {
this.receivedMsg = evt;
};
return new Observable(this.receivedMsg).share();
}
} …我无法弄清楚如何在rxjs中使用WebSocketSubjects v6.x
这是工作的HTML/JS v5.5.6.注释掉的代码是我试图让它在v6.x以下工作:
<html>
<head>
<!-- <script src="https://unpkg.com/@reactivex/rxjs@6.0.0/dist/global/rxjs.umd.js"></script> -->
<script src="https://unpkg.com/@reactivex/rxjs@5.5.6/dist/global/Rx.js"></script>
<script>
// const { WebSocketSubject } = rxjs.webSocket;
// const socket$ = WebSocketSubject.create('ws://localhost:8080');
const socket$ = Rx.Observable.webSocket('ws://localhost:8080');
socket$.subscribe(
(data) => console.log(data),
(err) => console.error(err),
() => console.warn('Completed!')
);
socket$.next(JSON.stringify({
event: 'events',
data: 'test',
}));
console.log('here')
</script>
</head>
<body></body>
</html>