Based on the input data I want to decide whether it is valid data or not and send the observable to the to end.
I have tried to create a observable and sequential subscribe.next() is working correctly but any error is coming then after that no statement is working.
const observable = new Observable((subscriber) => {
subscriber.next(1);
subscriber.next(2);
subscriber.error('failed this after second');
subscriber.next(3);
subscriber.next(4);
subscriber.complete();
});
observable.subscribe({
next(x) {
console.log('got value ' + x);
},
error(err) {
console.error('something wrong occurred: ' + err);
},
complete() {
console.log('done');
},
});
Current output:
Expected output
Here is the stackblitz link https://stackblitz.com/edit/zlksm5?devtoolsheight=50&file=index.ts
Emphasis mine.
If you want something that keeps going after it completes, you don't want an observable.
Just emit your error as a value. You could (for example) wrap your emissions in RxJS Notification objects so that you can materialize or dematerialize observables and keep a consistent API of emissions.
For example:
const observable = new Observable(subscriber => {
subscriber.next({ kind: "N", value: 1 });
subscriber.next({ kind: "N", value: 2 });
subscriber.next({ kind: "E", error: new Error("failed this after second") });
subscriber.next({ kind: "N", value: 3 });
subscriber.next({ kind: "N", value: 4 });
subscriber.next({ kind: "C" });
subscriber.complete();
});
/**** Alternative written more succinctly ****/
const observable = of(
{ kind: "N", value: 1 },
{ kind: "N", value: 2 },
{ kind: "E", error: new Error("failed this after second") },
{ kind: "N", value: 3 },
{ kind: "N", value: 4 },
{ kind: "C" }
);
observable.subscribe((notification) => {
switch (notification.kind) {
case "N":
console.log("got value ", notification.value);
break;
case "E":
console.error("something wrong occurred: ", notification.error.message);
break;
case "C":
console.log("done");
}
});
I don't think it's possible with an Observable, cause the observable will be closed if any error occurred.
I have tried couple of ways, putting it for reference
//interval(1000)
observable
//.pipe(switchMap(() => observable))
.pipe(
catchError((d) => {
observable.subscribe((nd) => {
console.log('--again--', nd);
});
return of(d);
})
//retry(1)
)
.subscribe({
next(x) {
console.log('got value ' + x);
},
error(err) {
console.error('something wrong occurred: ' + err);
},
complete() {
console.log('done');
},
});
// interval(1000)
// .pipe(switchMap(_ => observable))
// .pipe(
// catchError((d) => {return of(d)}),
// switchMap(_ => observable)
// )
// .subscribe(d => {
// console.log(d)
// })
For your use case it's better to use a behavior subject instead.