@@ -202,7 +202,7 @@ modify each individual request when using the `Do` method:
202202 function to process a streaming text response of type ` text/event-stream ` , where
203203 ` TextStreamCallback ` is ` func(TextStreamEvent) error ` . See below for more details.
204204* ` OptJsonStreamCallback(fn JsonStreamCallback) ` allows you to set a callback for JSON streaming
205- responses, where ` JsonStreamCallback ` is ` func(any ) error ` . See below for more details.
205+ responses, where ` JsonStreamCallback ` is ` func(json.RawMessage ) error ` . See below for more details.
206206
207207## Authentication
208208
@@ -334,7 +334,7 @@ func Callback(event client.TextStreamEvent) error {
334334
335335 // Decode the data into a JSON object
336336 var data map [string ]any
337- if err := event.Json (data); err != nil {
337+ if err := event.Json (& data); err != nil {
338338 return err
339339 }
340340
@@ -369,19 +369,149 @@ if err := stream.Decode(r, callback); err != nil {
369369}
370370```
371371
372- When the ` Accept ` type of the payload is ` text/event-stream ` or ` application/x-ndjson ` , the client
373- automatically adds ` Cache-Control: no-cache ` and ` X-Accel-Buffering: no ` request headers to
374- prevent intermediate proxies (including Nginx) from buffering the stream.
372+ ## JSON Streaming
375373
376- ## JSON Streaming Responses
374+ The client supports both one-way NDJSON response streaming and bi-directional NDJSON channels.
377375
378- The client decodes JSON streaming responses by passing a callback function to the ` OptJsonStreamCallback() ` option.
379- The callback with signature ` func(any) error ` is called for each JSON object in the stream, where the argument
380- is the same type as the object in the request.
376+ ### JSON Streaming Responses
377+
378+ For one-way JSON streaming responses, pass a callback function to the ` OptJsonStreamCallback() ` option.
379+ The callback with signature ` func(json.RawMessage) error ` is called for each JSON object in the stream.
380+ Decode the raw frame into the concrete type you expect.
381+
382+ ``` go
383+ package main
384+
385+ import (
386+ " encoding/json"
387+ " io"
388+ " log"
389+ " net/http"
390+
391+ client " github.com/mutablelogic/go-client"
392+ )
393+
394+ type Event struct {
395+ Value int ` json:"value"`
396+ }
397+
398+ func main () {
399+ c , err := client.New (client.OptEndpoint (" https://api.example.com"
400+ if err != nil {
401+ log.Fatal (err)
402+ }
403+
404+ err = c.Do (
405+ client.NewRequestEx (http.MethodGet , client.ContentTypeJsonStream ),
406+ nil ,
407+ client.OptPath (" events"
408+ client.OptNoTimeout (),
409+ client.OptJsonStreamCallback (func (v json.RawMessage ) error {
410+ var event Event
411+ if err := json.Unmarshal (v, &event); err != nil {
412+ return err
413+ }
414+ log.Printf (" value=%d " Value )
415+ if event.Value >= 100 {
416+ return io.EOF
417+ }
418+ return nil
419+ }),
420+ )
421+ if err != nil {
422+ log.Fatal (err)
423+ }
424+ }
425+ ```
381426
382427You can return an error from the callback to stop the stream and return the error, or return ` io.EOF ` to stop the stream
383428immediately and return success.
384429
430+ ### Bi-Directional JSON Streams
431+
432+ Use ` Client.Stream(ctx, opts...) ` to open a bi-directional JSON stream. The returned ` JSONStream `
433+ lets you send newline-delimited JSON request frames with ` Send ` , receive response frames with ` Recv ` ,
434+ close the outbound side with ` CloseSend ` , and tear down both sides with ` Close ` .
435+
436+ Blank response lines are treated as keep-alive heartbeats and ` Recv ` returns ` nil, nil ` for them.
437+
438+ ``` go
439+ package main
440+
441+ import (
442+ " context"
443+ " encoding/json"
444+ " io"
445+ " log"
446+ " time"
447+
448+ client " github.com/mutablelogic/go-client"
449+ )
450+
451+ type Reply struct {
452+ Echo string ` json:"echo"`
453+ }
454+
455+ func main () {
456+ ctx , cancel := context.WithCancel (context.Background ())
457+ defer cancel ()
458+
459+ // Create a new client
460+ c , err := client.New (client.OptEndpoint (" https://api.example.com"
461+ if err != nil {
462+ log.Fatal (err)
463+ }
464+
465+ // Create a bi-directional JSON stream to a server
466+ stream , err := c.Stream (ctx,client.OptPath (" session" " 1234" " channel" OptNoTimeout ())
467+ if err != nil {
468+ log.Fatal (err)
469+ }
470+ defer stream.Close ()
471+
472+ // Background goroutine to send JSON frames every 5 seconds until the context is cancelled
473+ go func () {
474+ ticker := time.NewTicker (5 * time.Second )
475+ defer ticker.Stop ()
476+
477+ for {
478+ select {
479+ case <- ctx.Done ():
480+ _ = stream.CloseSend ()
481+ return
482+ case <- ticker.C :
483+ frame := json.RawMessage (` {"text":"status"}`
484+ if err := stream.Send (frame); err != nil {
485+ log.Printf (" send error: %v "
486+ cancel ()
487+ return
488+ }
489+ }
490+ }
491+ }()
492+
493+ // Foreground loop to receive JSON frames until the stream is closed or an error occurs
494+ for {
495+ frame , err := stream.Recv ()
496+ if err == io.EOF {
497+ break
498+ }
499+ if err != nil {
500+ log.Fatal (err)
501+ }
502+ if len (frame) == 0 {
503+ continue // keep-alive heartbeat
504+ }
505+
506+ var reply Reply
507+ if err := json.Unmarshal (frame, &reply); err != nil {
508+ log.Fatal (err)
509+ }
510+ log.Printf (" reply=%s " Echo )
511+ }
512+ }
513+ ```
514+
385515## Transport Middleware
386516
387517The ` pkg/transport ` package provides composable ` http.RoundTripper ` middleware. All middleware
0 commit comments