Method Stream

internal/wsrelay/http.go:115–175

Stream executes a streaming HTTP request and returns channel with stream events.

(ctx context.Context, provider string, req *HTTPRequest)

// Stream executes a streaming HTTP request and returns channel with stream events.
func (m *Manager) Stream(ctx context.Context, provider string, req *HTTPRequest) (<-chan StreamEvent, error) {
	if req == nil {
		return nil, fmt.Errorf("wsrelay: request is nil")
	}
	msg := Message{ID: uuid.NewString(), Type: MessageTypeHTTPReq, Payload: encodeRequest(req)}
	respCh, err := m.Send(ctx, provider, msg)
	if err != nil {
		return nil, err
	}
	out := make(chan StreamEvent)
	go func() {
		defer close(out)
		send := func(ev StreamEvent) bool {
			if ctx == nil {
				out <- ev
				return true
			}
			select {
			case <-ctx.Done():
				return false
			case out <- ev:
				return true
			}
		}
		for {
			select {
			case <-ctx.Done():
				return
			case msg, ok := <-respCh:
				if !ok {
					_ = send(StreamEvent{Err: errors.New("wsrelay: stream closed")})
					return
				}
				switch msg.Type {
				case MessageTypeStreamStart:
					resp := decodeResponse(msg.Payload)
					if okSend := send(StreamEvent{Type: MessageTypeStreamStart, Status: resp.Status, Headers: resp.Headers}); !okSend {
						return
					}
				case MessageTypeStreamChunk:
					chunk := decodeChunk(msg.Payload)
					if okSend := send(StreamEvent{Type: MessageTypeStreamChunk, Payload: chunk}); !okSend {
						return
					}
				case MessageTypeStreamEnd:
					_ = send(StreamEvent{Type: MessageTypeStreamEnd})
					return
				case MessageTypeError:
					_ = send(StreamEvent{Type: MessageTypeError, Err: decodeError(msg.Payload)})
					return
				case MessageTypeHTTPResp:
					resp := decodeResponse(msg.Payload)
					_ = send(StreamEvent{Type: MessageTypeHTTPResp, Status: resp.Status, Headers: resp.Headers, Payload: resp.Body})
					return
				default:
				}
			}
		}