Streaming client

A generated client normally reads the entire response body and unmarshals it into a typed payload. To consume a stream you override that behavior — swap the consumer, and either buffer the whole thing or read it chunk-by-chunk. Two examples show both approaches.

Why the default doesn’t stream

The generated client hands the response body to a consumer, which does an io.Copy into the destination you pass. With the default JSONConsumer that copy blocks until the body is complete and then unmarshals — exactly what you don’t want for a stream. The fix is to install a ByteStreamConsumer for the response mime, so the bytes flow through untouched.

Non-blocking: consume chunks as they arrive

Override the consumer, then pass an io.Pipe writer as the destination. The client’s io.Copy writes into the pipe while a goroutine reads the other end — so bytes are processed the moment they arrive:

customized := httptransport.New("localhost:8000", "/", []string{"http"})
customized.Consumers[runtime.JSONMime] = runtime.ByteStreamConsumer()

countdowns := client.New(customized, nil)

reader, writer := io.Pipe()

scanner := bufio.NewScanner(reader)

Full source: stream-server/elapsed_client.go

A bufio.Scanner on the read end splits the stream on newlines and unmarshals each line independently. cancel() (via defer) tears down the request if the reader stops early:

wg.Add(1)
go func(wg *sync.WaitGroup) {
	defer wg.Done()
	defer cancel()

	// read response items line by line
	for scanner.Scan() {
		// each response item is JSON
		txt := scanner.Text()
		log.Printf("received countdown mark - raw: %s", txt)

		var mark models.Mark

		err := json.Unmarshal([]byte(txt), &mark)
		if err != nil {
			log.Printf("unmarshal error: %v", err)
			return
		}

		log.Printf("received countdown mark - remaining: %d", conv.Value(mark.Remains))
	}

	if err := scanner.Err(); err != nil {
		log.Printf("scanner err: %v", err)
	}

	log.Println("EOF")
}(&wg)

Full source: stream-server/elapsed_client.go

The request runs on the main goroutine, writing into the pipe the scanner drains; a context timeout bounds how long it will keep the connection open:

elapsed := operations.NewElapseParamsWithContext(queryCtx).WithLength(n)

_, err := countdowns.Operations.Elapse(elapsed, writer)

Full source: stream-server/elapsed_client.go

Blocking: buffer the whole stream

If you don’t need incremental processing, the simplest path is a destination that knows how to accept text/plain. Give a buffer an UnmarshalText method and the default consumer will fill it:

// Buffer knows how to UnmarshalText
type Buffer struct {
	*bytes.Buffer
}

// UnmarshalText handles text/plain
func (b *Buffer) UnmarshalText(text []byte) error {
	_, err := b.Write(text)
	return err
}

Full source: stream-client/jigsaw.go

Then pass it straight to the operation and read it once the call returns:

func chunkedBlocking(withChunks bool) error {
	c := client.New(customTransport(withChunks), nil).Operations

	// we just need to specify a buffer that knows how to UnmarshalText()
	buf := NewBuffer()
	_, err := c.Chunked(operations.NewChunkedParams(), buf)
	if err != nil {
		return err
	}

	data, err := io.ReadAll(buf)
	if err != nil {
		return err
	}

	log.Printf("result: %v", string(data))
	return nil
}

Full source: stream-client/jigsaw.go

The jigsaw.go example also has a non-blocking variant that installs transport.Consumers[runtime.TextMime] = runtime.ByteStreamConsumer() — the same technique as above, for a text/plain stream instead of newline-delimited JSON.

Choosing an approach

• Blocking + UnmarshalText — least code; fine when you can wait for the full response.
• Non-blocking + ByteStreamConsumer + io.Pipe — process items as they arrive, cancel early, bound with a context. Use this for long-lived or unbounded streams.

Related

• Streaming server — the countdown server elapsed_client.go consumes.