【发布时间】：2017-12-03 08:30:39
【问题描述】：
我有一个带有 rxjs WebsocketSubject 的 Angular 5 应用程序发送 jsonrpc 消息。
这是我的 sendRequest 函数
sendRequest(request: Request): Promise<Response> {
console.log(request);
this.socket.next(JSON.stringify(request));
return this.onResponse().filter((response: Response) => {
return response.id === request.id;
}).first().toPromise().then((response) => {
console.log(response);
if (response.error) {
console.log('error');
throw new RpcError(response.error);
}
return response;
});
}
我正在使用 first() 运算符来完成此过滤器订阅。但是 onResponse() 直接来自我的 WebsocketSubject，然后就会完成。
有什么方法可以解耦原始主体吗？
或者我应该创建一个新的 Observale.create(...)？
编写的 .filter 函数会发生什么。它会在任何地方持续存在，还是我必须在任何地方将其移除以防止过滤器调用持续存在？
编辑 1
同样使用它也无济于事。
sendRequest(request: Request): Promise<Response> {
console.log(request);
this.socket.next(JSON.stringify(request));
return new Promise<Response>((resolve, reject) => {
const responseSubscription = this.onResponse().filter((response: Response) => {
console.log('filter');
return response.id === request.id;
}).subscribe((response: Response) => {
// responseSubscription.unsubscribe();
resolve(response);
});
});
}
如果我执行取消订阅，整个 websocketSubject 就会关闭。不这样做会在每个请求上更准时地记录“过滤器”！！
编辑 2
这是我写的整个 websocketService
import {Injectable} from "@angular/core";
import {WebSocketSubject, WebSocketSubjectConfig} from "rxjs/observable/dom/WebSocketSubject";
import {MessageFactory, Notification, Request, Response, RpcError} from "../misc/jsonrpc";
import {ReplaySubject} from "rxjs/ReplaySubject";
import {Observable} from "rxjs/Observable";
import 'rxjs/add/operator/toPromise';
import 'rxjs/add/operator/filter';
import 'rxjs/add/operator/first';
import 'rxjs/add/observable/from';
export enum ConnectionState {
CONNECTED = "Connected",
CONNECTING = "Connecting",
CLOSING = "Closing",
DISCONNECTED = "Disconnected"
}
@Injectable()
export class WebsocketService {
private connectionState = new ReplaySubject<ConnectionState>(1);
private socket: WebSocketSubject<ArrayBuffer | Object>;
private config: WebSocketSubjectConfig;
constructor() {
console.log('ctor');
const protocol = location.protocol === 'https' ? 'wss' : 'ws';
const host = location.hostname;
const port = 3000; // location.port;
this.config = {
binaryType: "arraybuffer",
url: `${protocol}://${host}:${port}`,
openObserver: {
next: () => this.connectionState.next(ConnectionState.CONNECTED)
},
closingObserver: {
next: () => this.connectionState.next(ConnectionState.CLOSING)
},
closeObserver: {
next: () => this.connectionState.next(ConnectionState.DISCONNECTED)
},
resultSelector: (e: MessageEvent) => {
try {
if (e.data instanceof ArrayBuffer) {
return e.data;
} else {
return JSON.parse(e.data);
}
} catch (e) {
console.error(e);
return null;
}
}
};
this.connectionState.next(ConnectionState.CONNECTING);
this.socket = new WebSocketSubject(this.config);
this.connectionState.subscribe((state) => {
console.log(`WS state ${state}`);
});
}
onBinaryData(): Observable<ArrayBuffer> {
return this.socket.filter((message: any) => {
return message instanceof ArrayBuffer;
});
}
onMessageData(): Observable<Object> {
return this.socket.filter((message: any) => {
return !(message instanceof ArrayBuffer);
});
}
onResponse(): Observable<Response> {
return this.onMessageData().filter((message) => {
return MessageFactory.from(message).isResponse();
}).map((message): Response => {
return MessageFactory.from(message).toResponse();
});
}
sendRequest(request: Request): Promise<Response> {
console.log(request);
this.socket.next(JSON.stringify(request));
return new Promise<Response>((resolve, reject) => {
const responseSubscription = this.onResponse().filter((response: Response) => {
console.log('filter');
return response.id === request.id;
}).subscribe((response: Response) => {
responseSubscription.unsubscribe();
resolve(response);
});
});
}
sendNotification(notification: Notification): void {
this.socket.next(JSON.stringify(notification));
}
}
结果在我的日志中
Using Angular 5.0.2
websocket.service.ts:27 ctor
websocket.service.ts:69 WS state Connecting
core.js:3565 Angular is running in the development mode. Call enableProdMode() to enable the production mode.
websocket.service.ts:96 Request {jsonrpc: "2.0", id: "b042005c-5fbf-5ffc-fbd1-df68fae5882e", method: "appointment_list_get", params: undefined}
websocket.service.ts:69 WS state Connected
websocket.service.ts:103 filter
websocket.service.ts:69 WS state Disconnected
我需要找到一种方法，以某种方式将我的过滤器与原始流分离。
【问题讨论】：
-
为什么还要使用
first().toPromise().then？像this.onResponse().filter(...).subscribe(...);这样订阅不是更容易吗？ -
是的，但据我了解（也许我错了）我必须清理那个订阅，我期待一个单一的结果。所以使用 Promise 应该没问题。但是 .toPromise 只能在 Observalbe 完成时解决
-
那你现在的做法有什么问题？
-
.first() 完成整个流。所以我的 websocketSubject 完成了，因为 onResponse() 从我的 websocketSubject 返回了一个过滤的 Observable。所以我的 websocket 在每次响应后都会关闭
-
完成链并不能完成
this.onResponse()中的主题（或其他主题）。从您在编辑中显示的内容来看，问题将出在其他地方，而不是在完成 RxJS 链中。