Feat custom request

.circleci/config.yml

@@ -3,12 +3,12 @@ version: 2.1
 jobs:
   test:
     docker:
-      - image: golang:1.17
+      - image: golang:1.20
     steps:
       - checkout
       - run: make test
 
 workflows:
   build_and_test:
     jobs:
-      - test
+      - test

README.md

@@ -139,6 +139,45 @@ func main() {
 }
 ```
 
+If you want use this lib for openai sse client, you can follow this example:
+
+```go
+apiKey := "sk-***"
+log.SetLevel(log.Debug)
+ctx := context.TODO()
+client := sse.NewClient("https://api.openai.com/v1/chat/completions", func(c *sse.Client) {
+    c.IsDebug = false
+    // make any request, if you set MakeRequest then sse lib will not modify your request
+    c.MakeRequest = func() (*http.Request, error) {
+        buff := []byte(`{
+            "model":"gpt-3.5-turbo",
+            "stream": true,
+            "messages":[
+                {
+                    "role":"system",
+                    "content":"You are a helpful assistant."
+                },
+                {
+                    "role":"user",
+                    "content":"Hello!"
+                }
+            ]
+        }`)
+        req, err := http.NewRequest(http.MethodPost, "https://api.openai.com/v1/chat/completions", bytes.NewBuffer(buff))
+        req.Header["Authorization"] = []string{"Bearer " + apiKey}
+        req.Header["Content-Type"] = []string{"application/json"}
+        return req, err
+    }
+})
+err := client.SubscribeWithContext(ctx, "", func(msg *sse.Event) {
+    log.Infof("receive msg: %+v", msg)
+})
+if err != nil {
+    panic(err)
+}
+
+```
+
 #### HTTP client parameters
 
 To add additional parameters to the http client, such as disabling ssl verification for self signed certs, you can override the http client or update its options:

client.go

@@ -11,7 +11,9 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"log"
 	"net/http"
+	"net/http/httputil"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -38,6 +40,8 @@ type ConnCallback func(c *Client)
 // ResponseValidator validates a response
 type ResponseValidator func(c *Client, resp *http.Response) error
 
+type MakeRequest func() (*http.Request, error)
+
 // Client handles an incoming server stream
 type Client struct {
 	Retry             time.Time
@@ -55,6 +59,8 @@ type Client struct {
 	mu                sync.Mutex
 	EncodingBase64    bool
 	Connected         bool
+	IsDebug           bool
+	MakeRequest       MakeRequest
 }
 
 // NewClient creates a new client
@@ -289,34 +295,49 @@ func (c *Client) OnConnect(fn ConnCallback) {
 }
 
 func (c *Client) request(ctx context.Context, stream string) (*http.Response, error) {
-	req, err := http.NewRequest("GET", c.URL, nil)
+	var (
+		req *http.Request
+		err error
+	)
+	if c.MakeRequest != nil {
+		req, err = c.MakeRequest()
+	} else {
+		req, err = http.NewRequest("GET", c.URL, nil)
+		// Setup request, specify stream to connect to
+		if stream != "" {
+			query := req.URL.Query()
+			query.Add("stream", stream)
+			req.URL.RawQuery = query.Encode()
+		}
+
+		req.Header.Set("Cache-Control", "no-cache")
+		req.Header.Set("Accept", "text/event-stream")
+		req.Header.Set("Connection", "keep-alive")
+
+		lastID, exists := c.LastEventID.Load().([]byte)
+		if exists && lastID != nil {
+			req.Header.Set("Last-Event-ID", string(lastID))
+		}
+
+		// Add user specified headers
+		for k, v := range c.Headers {
+			req.Header.Set(k, v)
+		}
+	}
 	if err != nil {
 		return nil, err
 	}
 	req = req.WithContext(ctx)
-
-	// Setup request, specify stream to connect to
-	if stream != "" {
-		query := req.URL.Query()
-		query.Add("stream", stream)
-		req.URL.RawQuery = query.Encode()
+	if c.IsDebug {
+		reqBuff, _ := httputil.DumpRequest(req, true)
+		log.Printf("%s", string(reqBuff))
 	}
-
-	req.Header.Set("Cache-Control", "no-cache")
-	req.Header.Set("Accept", "text/event-stream")
-	req.Header.Set("Connection", "keep-alive")
-
-	lastID, exists := c.LastEventID.Load().([]byte)
-	if exists && lastID != nil {
-		req.Header.Set("Last-Event-ID", string(lastID))
+	resp, err := c.Connection.Do(req)
+	if c.IsDebug {
+		respBuff, _ := httputil.DumpResponse(resp, true)
+		log.Printf("%s", string(respBuff))
 	}
-
-	// Add user specified headers
-	for k, v := range c.Headers {
-		req.Header.Set(k, v)
-	}
-
-	return c.Connection.Do(req)
+	return resp, err
 }
 
 func (c *Client) processEvent(msg []byte) (event *Event, err error) {

go.mod

@@ -1,9 +1,15 @@
 module github.com/r3labs/sse/v2
 
-go 1.13
+go 1.20
 
 require (
 	github.com/stretchr/testify v1.7.0
-	golang.org/x/net v0.0.0-20191116160921-f9c825593386 // indirect
 	gopkg.in/cenkalti/backoff.v1 v1.1.0
 )
+
+require (
+	github.com/davecgh/go-spew v1.1.0 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	golang.org/x/net v0.0.0-20191116160921-f9c825593386 // indirect
+	gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c // indirect
+)