# Fx.observe

Observes the values of an `Fx` stream using a callback function.
The callback can return `void` or an `Effect` which will be executed for each value.

## Signatures

```ts
export declare const observe: {
    <A, E2 = never, R2 = never>(f: (value: A) => void | Effect<unknown, E2, R2>): <E, R>(fx: Fx<A, E, R>) => Effect<unknown, E | E2, R | R2>;
    <A, E, R, E2 = never, R2 = never>(fx: Fx<A, E, R>, f: (value: A) => void | Effect<unknown, E2, R2>): Effect<unknown, E | E2, R | R2>;
};
```

## Why

`observe` is the primary Effect boundary for performing work for each pushed value.
A plain callback is lifted to Effect, while an Effect callback contributes its own
typed failures and services to the result.

## Ownership and lifetime

Observation starts only when the returned Effect runs. That Effect owns the source
run and completes when the source completes. Interruption interrupts the internal
fiber and source cleanup. Callback invocation follows the producer's delivery
behavior; `observe` adds no buffer or concurrency of its own. A source cause or a
callback failure fails the returned Effect as `E | E2` and stops further
deliveries. The producer fiber finishes cleanup before observation settles;
resources acquired in the caller's Scope retain that Scope's lifetime.

## Examples

```ts
import { Effect, Ref } from "effect"
import { fromIterable, observe } from "@typed/fx/Fx"

const program = Effect.gen(function* () {
  const total = yield* Ref.make(0)
  yield* observe(fromIterable([1, 2, 3]), (value) => Ref.update(total, (n) => n + value))
  return yield* Ref.get(total)
})
```