Testing and Debugging RxJS Streams

Reactive programming with RxJS can be powerful, but debugging asynchronous data streams can get tricky. This notebook provides practical techniques for testing and debugging RxJS streams to help you troubleshoot issues and write reliable tests.

[1]:

$ npm install rxjs

added 2 packages in 2s

28 packages are looking for funding
  run `npm fund` for details

[2]:

RxJS loaded successfully

[M]:

1. Basic Debugging Techniques

Using tap() for Logging

The tap() operator is your best friend for debugging. It lets you peek into the stream without changing it.

[3]:

Debugging with tap():
Original value: 1
After map: 10
Original value: 2
After map: 20
Original value: 3
After map: 30
After filter: 30
Final value: 30
Original value: 4
After map: 40
After filter: 40
Final value: 40
Original value: 5
After map: 50
After filter: 50
Final value: 50
Stream completed

[M]:

Creating a Custom Debug Operator

Let's create a more powerful debug operator to consistently log stream values with context.

[4]:

Using custom debug operator:
[Source] Next: {"id":1,"name":"Alice"}
[After uppercase] Next: {"id":1,"name":"ALICE"}
[Source] Next: {"id":2,"name":"Bob"}
[After uppercase] Next: {"id":2,"name":"BOB"}
[After filter] Next: {"id":2,"name":"BOB"}
[Source] Next: {"id":3,"name":"Charlie"}
[After uppercase] Next: {"id":3,"name":"CHARLIE"}
[After filter] Next: {"id":3,"name":"CHARLIE"}
[Source] Complete
[After uppercase] Complete
[After filter] Complete

[M]:

2. Tracking Subscription Lifecycle

Tracking when subscriptions are created and disposed is critical for debugging memory leaks.

[5]:

Tracking subscription: Timer stream
Subscribing to Timer stream...
Timer stream emitted: 0
Timer stream emitted: 1
Timer stream emitted: 2
Stream Timer stream finalized

[M]:

3. Debugging Errors in RxJS Streams

Error handling is a critical part of working with RxJS.

[6]:

Error handling demo:
1. Without error handling:
[No error handling] Next: 1
Received: 1
[No error handling] Next: 2
Received: 2
[No error handling] Next: 3
Received: 3
[No error handling] Error: Error: Simulated error!
Error caught in subscriber: Simulated error!

2. With catchError for recovery:
[Before catchError] Next: 1
[After catchError] Next: 1
[Before catchError] Next: 2
[After catchError] Next: 2
[Before catchError] Next: 3
[After catchError] Next: 3
[Before catchError] Error: Error: Simulated error!
Error caught by operator: Simulated error!
[After catchError] Next: "Fallback value"
[After catchError] Complete
Stream completed with recovery

[M]:

4. Marble Testing

Marble testing is a powerful technique for testing RxJS streams. It uses ASCII diagrams to represent stream events over time.

[7]:

Basic marble testing example:
Actual:
 [
  {
    "frame": 2,
    "notification": {
      "kind": "N",
      "value": 10
    }
  },
  {
    "frame": 5,
    "notification": {
      "kind": "N",
      "value": 20
    }
  },
  {
    "frame": 8,
    "notification": {
      "kind": "C"
    }
  }
] 
Expected:
 [
  {
    "frame": 2,
    "notification": {
      "kind": "N",
      "value": 10
    }
  },
  {
    "frame": 5,
    "notification": {
      "kind": "N",
      "value": 20
    }
  },
  {
    "frame": 8,
    "notification": {
      "kind": "C"
    }
  }
] 
Test passed: true

[8]:

Testing async operators with marble diagrams:
Actual:
 [
  {
    "frame": 15,
    "notification": {
      "kind": "N",
      "value": 3
    }
  },
  {
    "frame": 18,
    "notification": {
      "kind": "C"
    }
  }
] 
Expected:
 [
  {
    "frame": 3,
    "notification": {
      "kind": "N",
      "value": 2
    }
  },
  {
    "frame": 5,
    "notification": {
      "kind": "N",
      "value": 3
    }
  },
  {
    "frame": 10,
    "notification": {
      "kind": "C"
    }
  }
] 
Test passed: false

[M]:

5. Memory Leak Detection

Memory leaks are a common issue with RxJS. Let's look at how to detect and fix them.

[9]:

Memory leak detection:
Creating a potential memory leak...

Creating a properly managed stream...
Leak emitted: 0
Safe stream emitted: 0
Leak emitted: 1
Safe stream emitted: 1

Cleaning up both streams...
Stream properly cleaned up

[M]:

6. Best Practices for Testing and Debugging

Add debug operators strategically:
- Place tap() calls before and after complex operators
- Use custom debug operators for consistent logging
- Remove or disable debug code in production
Always handle errors:
- Use catchError for recovery
- Consider retry strategies for transient failures
- Log errors with helpful context
Prevent memory leaks:
- Always unsubscribe from long-lived Observables
- Use the takeUntil(destroy$) pattern in components
- Use finalize() to verify cleanup
Write testable streams:
- Extract complex stream logic into functions
- Use dependency injection for external services
- Design with testing in mind
Use TestScheduler for time-based testing:
- TestScheduler uses virtual time, making tests fast and deterministic
- Define input streams with marble diagrams like '--a--b--|'
- Assert expected output with expectObservable(result$).toBe('--x--y--|')
Avoid nested subscriptions:
- Nested subscriptions are hard to track and often lead to memory leaks
- Use flattening operators like switchMap, mergeMap instead
Test edge cases:
- Empty streams
- Error cases
- Race conditions with multiple sources