# 223. Reactive Programming in Python

**🔹 1. Installing RxPY**

```sh
pip install rx
```

✅ **Fix:** Install the RxPY library before using it.

***

**🔹 2. Simple Observable and Observer Example**

```python
import rx
from rx import operators as ops

def print_value(x):
    print(x)

observable = rx.from_([1, 2, 3, 4, 5])
observable.subscribe(print_value)
```

✅ **Fix:** `rx.from_()` creates an **observable** and `subscribe()` attaches **observer** to it.

***

**🔹 3. Using `map` Operator to Transform Data**

```python
import rx
from rx import operators as ops

observable = rx.from_([1, 2, 3, 4, 5])
observable.pipe(
    ops.map(lambda x: x * 2)
).subscribe(lambda x: print(f"Transformed: {x}"))
```

✅ **Fix:** `map()` is used to **transform** items in the observable stream.

***

**🔹 4. Using `filter` Operator to Filter Data**

```python
import rx
from rx import operators as ops

observable = rx.from_([1, 2, 3, 4, 5, 6])
observable.pipe(
    ops.filter(lambda x: x % 2 == 0)
).subscribe(lambda x: print(f"Even number: {x}"))
```

✅ **Fix:** `filter()` **filters** the data based on a condition.

***

**🔹 5. Combining Multiple Observables with `merge`**

```python
import rx
from rx import operators as ops

observable1 = rx.from_([1, 2, 3])
observable2 = rx.from_([4, 5, 6])

rx.merge(observable1, observable2).subscribe(lambda x: print(f"Received: {x}"))
```

✅ **Fix:** `merge()` **combines** multiple observables into a single stream.

***

**🔹 6. Handling Errors with `on_error`**

```python
import rx
from rx import operators as ops

def error_func(x):
    if x == 3:
        raise Exception("Error at 3!")
    return x

observable = rx.from_([1, 2, 3, 4, 5]).pipe(
    ops.map(error_func)
)

observable.subscribe(
    on_next=lambda x: print(f"Received: {x}"),
    on_error=lambda e: print(f"Error: {e}")
)
```

✅ **Fix:** `on_error()` allows **handling** errors in reactive streams.

***

**🔹 7. Using `concat` for Sequential Execution**

```python
import rx
from rx import operators as ops

observable1 = rx.from_([1, 2])
observable2 = rx.from_([3, 4])

rx.concat(observable1, observable2).subscribe(lambda x: print(f"Received: {x}"))
```

✅ **Fix:** `concat()` ensures **sequential execution** of multiple observables.

***

**🔹 8. Delayed Execution with `interval`**

```python
import rx
from rx import operators as ops
import time

observable = rx.interval(1.0)  # Emits every second
observable.subscribe(lambda x: print(f"Tick {x}"))
time.sleep(5)  # Sleep for 5 seconds to observe the emitted ticks
```

✅ **Fix:** `interval()` emits values at fixed **time intervals**.

***

**🔹 9. Debouncing with `debounce`**

```python
import rx
from rx import operators as ops
import time

observable = rx.from_([1, 2, 3, 4, 5]).pipe(
    ops.debounce(lambda x: rx.timer(0.5))  # Wait 0.5 seconds after last event
)

observable.subscribe(lambda x: print(f"Received: {x}"))
```

✅ **Fix:** `debounce()` **delays** emissions until a certain time has passed.

***

**🔹 10. Using `zip` to Combine Observables**

```python
import rx
from rx import operators as ops

observable1 = rx.from_([1, 2, 3])
observable2 = rx.from_([4, 5, 6])

rx.zip(observable1, observable2).subscribe(lambda x: print(f"Received: {x}"))
```

✅ **Fix:** `zip()` **pairs up** values from multiple observables.

***

#### **🚀 Summary: Why Use RxPY?**

| Feature                   | RxPY                          |
| ------------------------- | ----------------------------- |
| **Reactive Programming**  | ✅ Handle asynchronous events  |
| **Combining Observables** | ✅ `merge`, `zip`, `concat`    |
| **Transformations**       | ✅ `map`, `filter`, `debounce` |
| **Error Handling**        | ✅ `on_error`                  |
| **Scheduled Emissions**   | ✅ `interval`, `timer`         |

***

#### **🚀 When to Use RxPY?**

| Scenario                                           | Use RxPY? |
| -------------------------------------------------- | --------- |
| **Handling real-time events (like UI updates)**    | ✅ Yes     |
| **Managing complex asynchronous workflows**        | ✅ Yes     |
| **Building event-driven systems or microservices** | ✅ Yes     |
| **Streaming data processing**                      | ✅ Yes     |

***
