草庐IT

javascript - 如何在 RxJS 中将可变长度数据包流转换为固定长度数据包?

我是RxJS的新手。给定以下流['foo','bar','b','az12','3','4567']我想将其转换为固定大小的数据包(比如3个字节)+余数['foo','ba','rb','az','123','456','7']在现实生活中,它实际上是一个二进制数据缓冲区。我想知道惯用的RxJS方法是什么。我发现的简单方法是:from(['foo','bar','b','az12','3','4567']).pipe(Rx.concatMap(v=>from(v)),Rx.bufferCount(3),Rx.map(v=>v.join(''))).subscribe(v=>consol

javascript - 合并两个 RxJS 流(基于同步)

我有两个RxJS主题,比如说a和b我需要以某种方式组合它们。someComboOfAandB.subscribe({aVal,bVal}=>console.log("value:",aVal,bVal));我想将它们结合起来,这样如果a和b同步更新,那么值会一起传递:a.next(1);//somecodeb.next(2)//atendofsynchronouscode/frame://value:12但是,如果只更新一个值,仍然会同时推送更新,同时推送具有两个新值的更新:a.next(5)//atendofsynchronouscode/frame://value:52这可能吗?如

javascript - Angular HttpInterceptor : How to use RxJS for multiple conditions

这是AuthInterceptor:@Injectable()exportclassAuthInterceptorimplementsHttpInterceptor{constructor(privateauthService:AuthService){}intercept(req:HttpRequest,next:HttpHandler):Observable>{constToken=this.authService.getToken();if(!Token){returnnext.handle(req);}//RefreshTokenfirstif(Token.expiresRef

javascript - Rxjs angular 6/7 mergeMap 延迟 http 请求

我想使用此代码发送请求(我也尝试过forkJoin),但调用之间有延迟:duplicateElement(id:string):Observable{returnthis.http.get({routeName:'route_name',params:{id}});}duplicateElements(ids:string[]):Observable{returnfrom(ids).pipe(mergeMap(id=>this.duplicateElement(id).pipe(delay(1000))));}但是.pipe(delay(1000)没有按照我的预期工作:在1000mls

javascript - 模拟 delay() RxJS 与 Jest

例如,是否有简单的方法在一个带有假时间的可观察对象中模拟RxJS的delay()方法?我有这个方法:register(user){returnthis._checkLog(user).delay(500).flatMap(...)}当我删除delay()方法时,我对_register()的测试全部成功。 最佳答案 RxJSv6对于这样的RxJSv6代码:代码.jsimport{of}from'rxjs';import{delay}from'rxjs/operators';exportconstexample=of('hello').

javascript - 如何从 `Observable` 导入 `Rx`(非 Angular )

我正在使用SystemJS将我的es2015项目加载到浏览器中。这是我做的import{Observable}from'rxjs/Rx';constbutton=document.querySelector('button');conststart$=Observable.fromEvent(button,'click');在这种情况下,Observable是undefined。所以我尝试了importObservablefrom'rxjs/Observable';在这种情况下,Observable是一个对象,但是Observable.fromEvent是undefined(它似乎是一

javascript - 使用 RxJS switchMap 仅取消订阅具有相同请求 URL/操作负载的流(redux-observable 史诗)

我有一个界面,用户可以在其中触发对同一端点但具有不同参数(在本例中为UUID)的调用。到目前为止,我一直很享受switchMap的行为,每当我发送一个具有相同类型的新redux操作时取消我的飞行中的http请求,在这种情况下我仍然想要这种行为,但是仅如果新操作请求的UUID(操作对象的一部分)与已经在进行中的UUID相同。我不太确定正确的方法。例如,在一次分派(dispatch)多个Action后,我希望所有具有唯一ID的Action都完成,但那些重复现有但尚未完成的ID会取消之前的请求并取而代之。例如:store.dispatch({type:"GET_SOME_DATA",uuid

javascript - Observable.Interval 的 RxJS 多重订阅

对于RxJS是否有类似以下解决方案的解决方案?Isitpossibletoinvokesubscribers'sOnNextsondifferentthreadsinRx?附言我的第一个天真的方法(在CoffeeScript中)显然失败了:hObs=Rx.Observable.interval(35000).startWith(-1).select(moment().format("DMMMM,HH:mm:ss")).publish()hObs.subscribe((x)->console.log(x))hObs.connect()hObs.subscribe((x)->console

javascript - RxJS:具有中止功能的生产者-消费者

我在RxJS中遇到了一个特殊的生产者消费者问题:生产者缓慢地生产元素。消费者正在请求元素,通常必须等待生产者。这可以通过压缩生产者和请求流来实现:varproduce=getProduceStream();varrequest=getRequestStream();varconsume=Rx.Observable.zipArray(produce,request).pluck(0);有时请求会被中止。生成的元素应该只在未中止的请求后使用:produce:-------------p1-------------------------p2--------->request:--r1---

javascript - RxJS:如何让一个 Observer 处理多个 Observables?

我正在使用调用我实现的函数的框架。我希望将此函数的参数转换为Observable,并通过一系列Observers发送。我以为我可以为此使用Subject,但它的行为并不像我预期的那样。为了澄清,我有类似下面的代码。我认为下面的Option1会起作用,但到目前为止,我正在接受Option2,这看起来一点也不符合习惯。vareventSubject=newRx.Subject();varresultSource=eventSubject.map(processEvent);varsubscription=resultSource.subscribe(function(event){cons