Method StreamChat

codersdk/chats.go:2833–2902

StreamChat streams chat updates in real time. The returned channel includes initial snapshot events first, followed by live updates. Callers must close the returned io.Closer to release the websocket connection when done.

(ctx context.Context, chatID uuid.UUID, opts *StreamChatOptions)

// live updates. Callers must close the returned io.Closer to release the
// websocket connection when done.
func (c *ExperimentalClient) StreamChat(ctx context.Context, chatID uuid.UUID, opts *StreamChatOptions) (<-chan ChatStreamEvent, io.Closer, error) {
	path := fmt.Sprintf("/api/experimental/chats/%s/stream", chatID)
	if opts != nil && opts.AfterID != nil {
		path += fmt.Sprintf("?after_id=%d", *opts.AfterID)
	}

	conn, err := c.Dial(
		ctx,
		path,
		&websocket.DialOptions{CompressionMode: websocket.CompressionDisabled},
	)
	if err != nil {
		return nil, nil, err
	}
	conn.SetReadLimit(1 << 22) // 4MiB

	streamCtx, streamCancel := context.WithCancel(ctx)
	events := make(chan ChatStreamEvent, 128)

	send := func(event ChatStreamEvent) bool {
		if event.ChatID == uuid.Nil {
			event.ChatID = chatID
		}
		select {
		case <-streamCtx.Done():
			return false
		case events <- event:
			return true
		}
	}

	go func() {
		defer close(events)
		defer streamCancel()
		defer func() {
			_ = conn.Close(websocket.StatusNormalClosure, "")
		}()

		for {
			var batch []ChatStreamEvent
			if err := wsjson.Read(streamCtx, conn, &batch); err != nil {
				if streamCtx.Err() != nil {
					return
				}
				switch websocket.CloseStatus(err) {
				case websocket.StatusNormalClosure, websocket.StatusGoingAway:
					return
				}
				_ = send(ChatStreamEvent{
					Type: ChatStreamEventTypeError,
					Error: &ChatError{
						Message: fmt.Sprintf("read chat stream: %v", err),
					},
				})
				return
			}

			for _, event := range batch {