111

Author: lvan100

gen/generator/golang/client.go

@@ -42,55 +42,51 @@ type Message struct {
 	Err  error
 }
 
+type IStream interface {
+	Send(msg Message) bool
+}
+
 // SSEEvent represents an SSE event.
-type SSEEvent[T any] struct {
+type SSEEvent struct {
 	ID    string
-	Data  T
+	Data  string
 	Event string
 	Retry int
 }
 
-type IStream interface {
-	Send(msg Message) bool
-}
-
 // Stream manages streaming data asynchronously from an HTTP response.
 // It supports safe concurrent use and can be closed idempotently.
-type Stream[T any] struct {
+type Stream struct {
 	msgs   chan Message
 	curr   Message
 	closed atomic.Bool
 	done   chan struct{}
 }
 
 // NewStream creates and initializes a new Stream instance.
-func NewStream[T any]() *Stream[T] {
-	return &Stream[T]{
+func NewStream() *Stream {
+	return &Stream{
 		msgs: make(chan Message),
 		done: make(chan struct{}),
 	}
 }
 
 // Event unmarshals the current data item into an SSEEvent.
-func (s *Stream[T]) Event() (SSEEvent[T], error) {
-	var e SSEEvent[T]
-	err := json.Unmarshal([]byte(s.curr.Data), &e)
-	if err != nil {
-		return SSEEvent[T]{}, err
-	}
+func (s *Stream) Event() (SSEEvent, error) {
+	var e SSEEvent
 	return e, nil
 }
 
 // Error returns the last error encountered by the stream.
-func (s *Stream[T]) Error() error {
+func (s *Stream) Error() error {
 	return s.curr.Err
 }
 
 // Next waits for the next data item from the stream,
 // honoring the provided context and optional timeout.
 // Returns true if a new data item is successfully received,
 // or false if the stream is closed, the context is done, or an error occurs.
-func (s *Stream[T]) Next(ctx context.Context, timeout time.Duration) bool {
+func (s *Stream) Next(ctx context.Context, timeout time.Duration) bool {
 	if s.closed.Load() {
 		return false
 	}
@@ -125,7 +121,7 @@ func (s *Stream[T]) Next(ctx context.Context, timeout time.Duration) bool {
 
 // Send pushes a Message into the internal channel.
 // Returns false if the stream is closed or already done.
-func (s *Stream[T]) Send(msg Message) bool {
+func (s *Stream) Send(msg Message) bool {
 	if s.closed.Load() {
 		return false
 	}
@@ -139,7 +135,7 @@ func (s *Stream[T]) Send(msg Message) bool {
 
 // Close closes the Stream safely (idempotent).
 // It ensures that the internal channels are closed only once.
-func (s *Stream[T]) Close() {
+func (s *Stream) Close() {
 	if s.closed.CompareAndSwap(false, true) {
 		close(s.done)
 		close(s.msgs)
@@ -315,8 +311,8 @@ func JSONResponse[RespType any](r *http.Request, opts ...RequestOption) (*http.R
 
 // StreamResponse executes the given HTTP request using the provided HTTPClient,
 // and returns a Stream instance for streaming the response body.
-func StreamResponse[T any](r *http.Request, opts ...RequestOption) (*http.Response, *Stream[T], error) {
-	s := NewStream[T]()
+func StreamResponse(r *http.Request, opts ...RequestOption) (*http.Response, *Stream, error) {
+	s := NewStream()
 	resp, err := DefaultClient.Stream(r, buildMeta(opts), s)
 	if err != nil {
 		return nil, nil, err

lib/httputil/httputil.go

@@ -295,7 +295,7 @@ type StreamResponse struct {
 
 // Stream sends a POST request to the /v1/stream endpoint with the given request body.
 func (c *HelloClient) Stream(ctx context.Context, req *StreamRequest, opts ...httputil.RequestOption) (
-	*http.Response, *httputil.Stream[*StreamResponse], error) {
+	*http.Response, *httputil.Stream, error) {
 
 	path := "/v1/stream"
 	if s, err := req.QueryString(); err != nil {
@@ -321,7 +321,7 @@ func (c *HelloClient) Stream(ctx context.Context, req *StreamRequest, opts ...ht
 	opts = append(opts, httputil.WithTarget(c.ServiceName))
 	opts = append(opts, httputil.WithPath("/v1/hello"))
 	opts = append(opts, httputil.WithSchema("http"))
-	return httputil.StreamResponse[*StreamResponse](r, opts...)
+	return httputil.StreamResponse(r, opts...)
 }
 
 func TestHello(t *testing.T) {
@@ -394,8 +394,8 @@ func TestStream(t *testing.T) {
 		_ = r.Header.Write(os.Stdout)
 		fmt.Println()
 		for i := range 5 {
-			_, _ = w.Write([]byte(fmt.Sprintf("%d: ", i)))
-			_, _ = w.Write([]byte(`{"message": "hello world"}`))
+			_, _ = w.Write([]byte(fmt.Sprintf("id: %v\n", i)))
+			_, _ = w.Write([]byte(fmt.Sprintf("data: %s\n", `{"message": "hello world"}`)))
 			_, _ = w.Write([]byte("\n\n"))
 			w.(http.Flusher).Flush()
 			time.Sleep(time.Millisecond * 500)