# Streaming

## Stream JSON Lines

To stream JSON Lines, declare the return type and use `yield` to return the data.

```python
@app.get("/items/stream")
async def stream_items() -> AsyncIterable[Item]:
    for item in items:
        yield item
```

## Server-Sent Events (SSE)

To stream Server-Sent Events, use `response_class=EventSourceResponse` and `yield` items from the endpoint.

items = [
    Item(name="Plumbus", price=32.99),
    Item(name="Portal Gun", price=999.99),
    Item(name="Meeseeks Box", price=49.99),
]


@app.get("/items/stream", response_class=EventSourceResponse)
async def stream_items() -> AsyncIterable[ServerSentEvent]:
    yield ServerSentEvent(comment="stream of item updates")
    for i, item in enumerate(items):

    Item(name="Portal Gun", description="A portal opening device."),
    Item(name="Meeseeks Box", description="A box that summons a Meeseeks."),
]


@app.get("/items/stream")
async def stream_items() -> AsyncIterable[Item]:
    for item in items:
        yield item


@app.get("/items/stream-no-async")
def stream_items_no_async() -> Iterable[Item]:
    for item in items:
        yield item

			if !ok {
				return
			}

			if !send(key) {
				return
			}
		case <-doneCh:
			return
		}
	}
}

// StreamItems reads the keys from the store and replays the corresponding item to the target.
func StreamItems[I any](store Store[I], target Target, doneCh <-chan struct{}, logger logger) {
	go func() {
		keyCh := replayItems(store, doneCh, logger, target.Name())

		id:         event.TargetID{ID: id, Name: "nsq"},
		args:       args,
		loggerOnce: loggerOnce,
		store:      queueStore,
		quitCh:     make(chan struct{}),
	}

	if target.store != nil {
		store.StreamItems(target.store, target, target.quitCh, target.loggerOnce)
	}

	return target, nil

		id:         event.TargetID{ID: id, Name: "mqtt"},
		args:       args,
		store:      queueStore,
		quitCh:     make(chan struct{}),
		loggerOnce: loggerOnce,
	}

	if target.store != nil {
		store.StreamItems(target.store, target, target.quitCh, target.loggerOnce)
	}

	return target, nil

		case "http":
			target.addr += ":80"
		case "https":
			target.addr += ":443"
		default:
			return nil, errors.New("unsupported scheme")
		}
	}

	if target.store != nil {
		store.StreamItems(target.store, target, target.cancelCh, target.loggerOnce)
	}

	return target, nil

		return fmt.Errorf("unable to initialize the queue store of %s webhook: %w", h.Name(), err)
	}
	ctx, cancel := context.WithCancel(ctx)
	h.store = queueStore
	h.storeCtxCancel = cancel
	store.StreamItems(h.store, h, ctx.Done(), h.kconfig.LogOnce)
	return err
}

func (h *Target) startKafkaLogger() {
	h.logChMu.RLock()
	logCh := h.logCh
	if logCh != nil {
		// We are not allowed to add when logCh is nil

		args:       args,
		pool:       pool,
		store:      queueStore,
		loggerOnce: loggerOnce,
		quitCh:     make(chan struct{}),
	}

	if target.store != nil {
		store.StreamItems(target.store, target, target.quitCh, target.loggerOnce)
	}

	return target, nil

				Limit:         args.BatchSize,
				Log:           loggerOnce,
				Store:         queueStore,
				CommitTimeout: args.BatchCommitTimeout,
			})
		}
		store.StreamItems(target.store, target, target.quitCh, target.loggerOnce)
	}

	return target, nil
}

func isKafkaConnErr(err error) bool {
	// Sarama opens the circuit breaker after 3 consecutive connection failures.