Refactor streaming code and remove versioning (#65)

* Updated streaming code
* Removed version
* Updated client streadming with example
* Updatting stream
* Updated bitwarden
* Fixes
* Updated README

Author: djthorpe

README.md

@@ -402,7 +402,7 @@ func main() {
     }
 
     err = c.Do(
-        client.NewRequestEx(http.MethodGet, client.ContentTypeJsonStream),
+        client.NewRequestEx(http.MethodGet, "application/ndjson"),
         nil,
         client.OptPath("events"),
         client.OptNoTimeout(),
@@ -429,19 +429,26 @@ immediately and return success.
 
 ### Bi-Directional JSON Streams
 
-Use `Client.Stream(ctx, opts...)` to open a bi-directional JSON stream. The returned `JSONStream`
-lets you send newline-delimited JSON request frames with `Send`, receive response frames with `Recv`,
-close the outbound side with `CloseSend`, and tear down both sides with `Close`.
+Use `Client.Stream(ctx, callback, opts...)` to open a bi-directional NDJSON stream.
+The callback receives a `JSONStream`, which lets you send newline-delimited JSON
+request frames with `Send` and receive response frames from the channel returned
+by `Recv`.
 
-Blank response lines are treated as keep-alive heartbeats and `Recv` returns `nil, nil` for them.
+Returning from the callback closes the stream. Canceling the context passed to
+`Client.Stream` also closes the stream. Blank response lines are treated as
+keep-alive heartbeats and are delivered as `nil` frames on the receive channel.
+
+The receive side starts immediately when the stream opens. Callbacks should keep
+draining `Recv()` while the stream is active; if responses are not consumed and
+the internal receive buffer fills, the stream is canceled to avoid a full-duplex
+deadlock.
 
 ```go
 package main
 
 import (
     "context"
     "encoding/json"
-    "io"
     "log"
     "time"
 
@@ -462,52 +469,37 @@ func main() {
         log.Fatal(err)
     }
 
-    // Create a bi-directional JSON stream to a server
-    stream, err := c.Stream(ctx,client.OptPath("session", "1234", "channel"),client.OptNoTimeout())
-    if err != nil {
-        log.Fatal(err)
-    }
-    defer stream.Close()
-
-    // Background goroutine to send JSON frames every 5 seconds until the context is cancelled
-    go func() {
+    // Run the stream until the callback returns or the context is cancelled.
+    if err := c.Stream(ctx, func(ctx context.Context, stream client.JSONStream) error {
         ticker := time.NewTicker(5 * time.Second)
         defer ticker.Stop()
 
         for {
             select {
-            case <-ctx.Done():
-                _ = stream.CloseSend()
-                return
             case <-ticker.C:
                 frame := json.RawMessage(`{"text":"status"}`)
                 if err := stream.Send(frame); err != nil {
-                    log.Printf("send error: %v", err)
-                    cancel()
-                    return
+                    return err
+                }
+            case frame, ok := <-stream.Recv():
+                if !ok {
+                    return nil
+                }
+                if frame == nil {
+                    continue // keep-alive heartbeat
                 }
-            }
-        }
-    }()
-
-    // Foreground loop to receive JSON frames until the stream is closed or an error occurs
-    for {
-        frame, err := stream.Recv()
-        if err == io.EOF {
-            break
-        }
-        if err != nil {
-            log.Fatal(err)
-        }
-        if len(frame) == 0 {
-            continue // keep-alive heartbeat
-        }
 
-        var reply Reply
-        if err := json.Unmarshal(frame, &reply); err != nil {
-            log.Fatal(err)
+                var reply Reply
+                if err := json.Unmarshal(frame, &reply); err != nil {
+                    return err
+                }
+                log.Printf("reply=%s", reply.Echo)
+            case <-ctx.Done():
+                return ctx.Err()
+            }
         }
-        log.Printf("reply=%s", reply.Echo)
+    }, client.OptPath("session", "1234", "channel"), client.OptNoTimeout()); err != nil {
+        log.Fatal(err)
     }
 }
 ```
@@ -522,8 +514,8 @@ follows the same constructor pattern `New*(... , parent http.RoundTripper)` and
 ### Logging Transport
 
 `transport.NewLogging` logs every request and response to an `io.Writer`. When `verbose` is true
-the request and response bodies are also printed; `text/event-stream` bodies are printed
-line-by-line as events arrive rather than buffered:
+the request and response bodies are also printed; `text/event-stream` and NDJSON streaming bodies
+are printed line-by-line as events arrive rather than buffered:
 
 ```go
 import (

client.go

@@ -57,17 +57,9 @@ type JsonStreamCallback func(json.RawMessage) error
 // GLOBALS
 
 const (
-	DefaultTimeout            = time.Second * 30
-	DefaultUserAgent          = "github.com/mutablelogic/go-client"
-	PathSeparator             = "/"
-	ContentTypeAny            = types.ContentTypeAny
-	ContentTypeJson           = types.ContentTypeJSON
-	ContentTypeJsonStream     = "application/x-ndjson"
-	ContentTypeTextXml        = types.ContentTypeTextXml
-	ContentTypeApplicationXml = types.ContentTypeXML
-	ContentTypeTextPlain      = types.ContentTypeTextPlain
-	ContentTypeTextHTML       = "text/html"
-	ContentTypeBinary         = types.ContentTypeBinary
+	DefaultTimeout   = time.Second * 30
+	DefaultUserAgent = "github.com/mutablelogic/go-client"
+	PathSeparator    = "/"
 )
 
 ///////////////////////////////////////////////////////////////////////////////
@@ -200,19 +192,19 @@ func (client *Client) request(ctx context.Context, method, accept, mimetype stri
 	// Set the credentials and user agent
 	if body != nil {
 		if mimetype == "" {
-			mimetype = ContentTypeJson
+			mimetype = types.ContentTypeJSON
 		}
 		r.Header.Set("Content-Type", mimetype)
 	}
 	if accept != "" {
 		r.Header.Set("Accept", accept)
 	} else {
-		r.Header.Set("Accept", ContentTypeAny)
+		r.Header.Set("Accept", types.ContentTypeAny)
 	}
 	// For SSE or NDJSON streams, disable caching and Nginx proxy buffering so
 	// events are delivered immediately rather than held in intermediate buffers.
 	// Accept may be a comma-separated list so use Contains rather than ==.
-	if strings.Contains(accept, ContentTypeTextStream) || strings.Contains(accept, ContentTypeJsonStream) || strings.Contains(accept, types.ContentTypeJSONStream) {
+	if strings.Contains(accept, ContentTypeTextStream) || strings.Contains(accept, types.ContentTypeJSONStream) {
 		r.Header.Set("Cache-Control", "no-cache")
 		r.Header.Set("X-Accel-Buffering", "no")
 	}
@@ -361,7 +353,7 @@ func do(client *http.Client, req *http.Request, accept string, strict bool, out
 	// When in strict mode, check content type returned is as expected.
 	// Use 406 Not Acceptable since this is client-side validation that the
 	// server's response doesn't match our Accept header expectations.
-	if strict && (accept != "" && accept != ContentTypeAny) {
+	if strict && (accept != "" && accept != types.ContentTypeAny) {
 		if mimetype != accept {
 			return httpresponse.Err(http.StatusNotAcceptable).Withf("strict mode: expected %q, got %q", accept, mimetype)
 		}
@@ -389,7 +381,7 @@ func do(client *http.Client, req *http.Request, accept string, strict bool, out
 	}
 
 	switch {
-	case mimetype == ContentTypeJson || isJSONStreamContentType(mimetype):
+	case mimetype == types.ContentTypeJSON || mimetype == types.ContentTypeJSONStream:
 		dec := json.NewDecoder(response.Body)
 		if reqopts.jsonStreamCallback != nil {
 			for {
@@ -416,11 +408,11 @@ func do(client *http.Client, req *http.Request, accept string, strict bool, out
 				return err
 			}
 		}
-	case mimetype == ContentTypeTextXml || mimetype == ContentTypeApplicationXml:
+	case mimetype == types.ContentTypeTextXml || mimetype == types.ContentTypeXML:
 		if err := xml.NewDecoder(response.Body).Decode(out); err != nil {
 			return err
 		}
-	case mimetype == ContentTypeTextPlain:
+	case mimetype == types.ContentTypeTextPlain:
 		data, err := io.ReadAll(response.Body)
 		if err != nil {
 			return err
@@ -455,7 +447,7 @@ func do(client *http.Client, req *http.Request, accept string, strict bool, out
 func respContentType(resp *http.Response) (string, error) {
 	contenttype := resp.Header.Get("Content-Type")
 	if contenttype == "" {
-		return ContentTypeBinary, nil
+		return types.ContentTypeBinary, nil
 	}
 	if mimetype, err := types.ParseContentType(contenttype); err != nil {
 		return contenttype, httpresponse.Err(http.StatusUnsupportedMediaType).With(contenttype)

example/streaming/client.go

@@ -0,0 +1,52 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"math/rand"
+	"time"
+
+	// Packages
+	client "github.com/mutablelogic/go-client"
+)
+
+func runClient(ctx context.Context, listenAddr string) error {
+	// Create a client
+	c, err := client.New(client.OptEndpoint("http://" + listenAddr))
+	if err != nil {
+		return err
+	}
+
+	// Create a bi-directional stream
+	if err := c.Stream(ctx, func(ctx context.Context, stream client.JSONStream) error {
+		ticker := time.NewTimer(time.Second)
+		defer ticker.Stop()
+		var seq int
+		for {
+			select {
+			case <-ticker.C:
+				if err := stream.Send(Event{Message: fmt.Sprintf("client seq %d", seq)}.JSON()); err != nil {
+					return err
+				}
+				seq++
+				ticker.Reset(time.Second * time.Duration(rand.Int31n(8)))
+			case evt, ok := <-stream.Recv():
+				if !ok {
+					return nil
+				}
+				if e, err := NewEvent(evt); err != nil {
+					return err
+				} else {
+					fmt.Println("received event:", e)
+				}
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+		}
+	}); err != nil {
+		return err
+	}
+
+	// Create a ticker
+	return nil
+}

example/streaming/event.go

@@ -0,0 +1,41 @@
+package main
+
+import (
+	"encoding/json"
+
+	// Packages
+	"github.com/mutablelogic/go-server/pkg/types"
+)
+
+///////////////////////////////////////////////////////////////////////////////
+// TYPES
+
+type Event struct {
+	Message string `json:"message"`
+}
+
+///////////////////////////////////////////////////////////////////////////////
+// STRINGIFY
+
+func (e Event) String() string {
+	return types.Stringify(e)
+}
+
+///////////////////////////////////////////////////////////////////////////////
+// PUBLIC METHODS
+
+func NewEvent(data json.RawMessage) (Event, error) {
+	var evt Event
+	if err := json.Unmarshal(data, &evt); err != nil {
+		return Event{}, err
+	}
+	return evt, nil
+}
+
+func (e Event) JSON() json.RawMessage {
+	if data, err := json.Marshal(e); err != nil {
+		return nil
+	} else {
+		return data
+	}
+}

example/streaming/main.go

@@ -0,0 +1,58 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/signal"
+	"path/filepath"
+	"syscall"
+)
+
+///////////////////////////////////////////////////////////////////////////////
+// GLOBALS
+
+const (
+	ListenAdress = "localhost:8080"
+)
+
+///////////////////////////////////////////////////////////////////////////////
+// MAIN
+
+func main() {
+	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer cancel()
+	if client, err := isClient(); err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		os.Exit(1)
+	} else if client {
+		fmt.Fprintln(os.Stdout, "running client")
+		if err := runClient(ctx, ListenAdress); err != nil {
+			fmt.Fprintln(os.Stderr, err)
+			os.Exit(1)
+		}
+	} else {
+		fmt.Fprintln(os.Stdout, "running server")
+		if err := runServer(ctx, ListenAdress); err != nil {
+			fmt.Fprintln(os.Stderr, err)
+			os.Exit(1)
+		}
+	}
+}
+
+///////////////////////////////////////////////////////////////////////////////
+// PRIVATE METHODS
+
+func isClient() (bool, error) {
+	if len(os.Args) < 2 {
+		return false, fmt.Errorf("usage: %s [client|server]", filepath.Base(os.Args[0]))
+	}
+	switch os.Args[1] {
+	case "client":
+		return true, nil
+	case "server":
+		return false, nil
+	default:
+		return false, fmt.Errorf("invalid argument: %q", os.Args[1])
+	}
+}