Method Transmit

codersdk/agentsdk/agentsdk.go:885–922

Transmit will read from the given chan and send events for as long as: * the chan remains open * the context has not been canceled * not timed out * the connection to the receiver remains open

(ctx context.Context, reinitEvents <-chan ReinitializationEvent)

// * not timed out
// * the connection to the receiver remains open
func (s *SSEAgentReinitTransmitter) Transmit(ctx context.Context, reinitEvents <-chan ReinitializationEvent) error {
	select {
	case <-ctx.Done():
		return ctx.Err()
	default:
	}

	sseSendEvent, sseSenderClosed, err := httpapi.ServerSentEventSender(s.rw, s.r)
	if err != nil {
		return xerrors.Errorf("failed to create sse transmitter: %w", err)
	}

	defer func() {
		// Block returning until the ServerSentEventSender is closed
		// to avoid a race condition where we might write or flush to rw after the handler returns.
		<-sseSenderClosed
	}()

	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-sseSenderClosed:
			return ErrTransmissionTargetClosed
		case reinitEvent, ok := <-reinitEvents:
			if !ok {
				return ErrTransmissionSourceClosed
			}
			err := sseSendEvent(codersdk.ServerSentEvent{
				Type: codersdk.ServerSentEventTypeData,
				Data: reinitEvent,
			})
			if err != nil {
				return err
			}
		}
	}
}

func NewSSEAgentReinitReceiver(r io.ReadCloser) *SSEAgentReinitReceiver {
	return &SSEAgentReinitReceiver{r: r}