feat(api): add events streaming

Author: stainless-app[bot]

.stats.yml

@@ -1,2 +1,2 @@
 configured_endpoints: 106
-openapi_spec_url: https://storage.googleapis.com/stainless-sdk-openapi-specs/gitpod%2Fgitpod-2e9f8b8666b2fd4e346a3acbf81a2c82a6f3793e01bc146499708efaf0c250c5.yml
+openapi_spec_url: https://storage.googleapis.com/stainless-sdk-openapi-specs/gitpod%2Fgitpod-922f204ec36b8a84ae8f96e73923e92cb2044a14c6497d173f4b7110a090ac30.yml

event.go

@@ -14,6 +14,7 @@ import (
 	"github.com/stainless-sdks/gitpod-go/internal/param"
 	"github.com/stainless-sdks/gitpod-go/internal/requestconfig"
 	"github.com/stainless-sdks/gitpod-go/option"
+	"github.com/stainless-sdks/gitpod-go/packages/jsonl"
 )
 
 // EventService contains methods and other services that help with interacting with
@@ -51,18 +52,22 @@ func (r *EventService) List(ctx context.Context, params EventListParams, opts ..
 }
 
 // WatchEvents streams all requests events to the client
-func (r *EventService) Watch(ctx context.Context, params EventWatchParams, opts ...option.RequestOption) (res *EventWatchResponse, err error) {
+func (r *EventService) WatchStreaming(ctx context.Context, params EventWatchParams, opts ...option.RequestOption) (stream *jsonl.Stream[EventWatchResponse]) {
+	var (
+		raw *http.Response
+		err error
+	)
 	if params.ConnectProtocolVersion.Present {
 		opts = append(opts, option.WithHeader("Connect-Protocol-Version", fmt.Sprintf("%s", params.ConnectProtocolVersion)))
 	}
 	if params.ConnectTimeoutMs.Present {
 		opts = append(opts, option.WithHeader("Connect-Timeout-Ms", fmt.Sprintf("%s", params.ConnectTimeoutMs)))
 	}
 	opts = append(r.Options[:], opts...)
-	opts = append([]option.RequestOption{option.WithHeader("Accept", "application/connect+json")}, opts...)
+	opts = append([]option.RequestOption{option.WithHeader("Accept", "application/jsonl")}, opts...)
 	path := "gitpod.v1.EventService/WatchEvents"
-	err = requestconfig.ExecuteNewRequest(ctx, http.MethodPost, path, params, &res, opts...)
-	return
+	err = requestconfig.ExecuteNewRequest(ctx, http.MethodPost, path, params, &raw, opts...)
+	return jsonl.NewStream[EventWatchResponse](raw, err)
 }
 
 type EventListResponse struct {

event_test.go

@@ -42,31 +42,3 @@ func TestEventListWithOptionalParams(t *testing.T) {
 		t.Fatalf("err should be nil: %s", err.Error())
 	}
 }
-
-func TestEventWatchWithOptionalParams(t *testing.T) {
-	baseURL := "http://localhost:4010"
-	if envURL, ok := os.LookupEnv("TEST_API_BASE_URL"); ok {
-		baseURL = envURL
-	}
-	if !testutil.CheckTestServer(t, baseURL) {
-		return
-	}
-	client := gitpod.NewClient(
-		option.WithBaseURL(baseURL),
-		option.WithBearerToken("My Bearer Token"),
-	)
-	_, err := client.Events.Watch(context.TODO(), gitpod.EventWatchParams{
-		Body: gitpod.EventWatchParamsBodyEnvironmentScopeProducesEventsForTheEnvironmentItselfAllTasksTaskExecutionsAndServicesAssociatedWithThatEnvironment{
-			EnvironmentID: gitpod.F("environmentId"),
-		},
-		ConnectProtocolVersion: gitpod.F(gitpod.EventWatchParamsConnectProtocolVersion1),
-		ConnectTimeoutMs:       gitpod.F(0.000000),
-	})
-	if err != nil {
-		var apierr *gitpod.Error
-		if errors.As(err, &apierr) {
-			t.Log(string(apierr.DumpRequest(true)))
-		}
-		t.Fatalf("err should be nil: %s", err.Error())
-	}
-}

packages/jsonl/jsonl.go

@@ -0,0 +1,55 @@
+// File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
+
+package jsonl
+
+import (
+	"bufio"
+	"encoding/json"
+	"io"
+	"net/http"
+)
+
+type Stream[T any] struct {
+	rc  io.ReadCloser
+	scn *bufio.Scanner
+	cur T
+	err error
+}
+
+func NewStream[T any](res *http.Response, err error) *Stream[T] {
+	if res == nil || res.Body == nil {
+		return nil
+	}
+
+	return &Stream[T]{
+		rc:  res.Body,
+		scn: bufio.NewScanner(res.Body),
+		err: err,
+	}
+}
+
+func (s *Stream[T]) Next() bool {
+	if s.err != nil {
+		return false
+	}
+
+	if !s.scn.Scan() {
+		return false
+	}
+
+	line := s.scn.Bytes()
+	s.err = json.Unmarshal(line, &s.cur)
+	return s.err == nil
+}
+
+func (s *Stream[T]) Current() T {
+	return s.cur
+}
+
+func (s *Stream[T]) Err() error {
+	return s.err
+}
+
+func (s *Stream[T]) Close() error {
+	return s.rc.Close()
+}