make start events private

Author: emranbm

errors.go

@@ -2,5 +2,4 @@ package golangfuse
 
 import "errors"
 
-var AlreadyStartedErr = errors.New("already started")
 var AlreadyShutdownErr = errors.New("already shutdown")

langfuse.go

@@ -3,7 +3,6 @@ package golangfuse
 import (
 	"context"
 	"fmt"
-	"net/http"
 	"sync/atomic"
 	"time"
 
@@ -13,7 +12,6 @@ import (
 )
 
 type Langfuse interface {
-	StartSendingEvents(ctx context.Context, period time.Duration) error
 	// Shutdown stops the event buffer and cleans up resources.
 	// It also ensures that any pending events are sent before shutdown.
 	Shutdown(ctx context.Context) error
@@ -22,52 +20,46 @@ type Langfuse interface {
 }
 
 type langfuseImpl struct {
-	restClient             *resty.Client
-	eventBuffer            *eventBuffer
-	isSendingEventsStarted atomic.Bool
-	isShutdown             atomic.Bool
-	cancelFunc             context.CancelFunc
-	endpoint               string
-	promptLabel            string
+	restClient  *resty.Client
+	eventBuffer *eventBuffer
+	isShutdown  atomic.Bool
+	cancelFunc  context.CancelFunc
+	endpoint    string
+	promptLabel string
+	flushPeriod time.Duration
 }
 
-func New(endpoint, publicKey, secretKey string) Langfuse {
-	return NewWithHttpClient(http.DefaultClient, endpoint, publicKey, secretKey)
-}
-
-func NewWithHttpClient(httpClient *http.Client, endpoint, publicKey, secretKey string) Langfuse {
-	client := resty.NewWithClient(httpClient).SetBasicAuth(publicKey, secretKey)
+func New(
+	ctx context.Context,
+	endpoint,
+	publicKey,
+	secretKey string,
+	opts ...Option,
+) Langfuse {
 	c := &langfuseImpl{
-		restClient:  client,
+		restClient:  resty.New(),
 		endpoint:    endpoint,
-		promptLabel: "production", // TODO: use option pattern to override this default if needed
+		promptLabel: "production",
+		flushPeriod: 1 * time.Second,
+	}
+	for _, opt := range opts {
+		opt(c)
 	}
+	c.restClient = c.restClient.SetBasicAuth(publicKey, secretKey)
 	c.eventBuffer = newEventBufferer(c.sendEvents)
+	c.startSendingEvents(ctx, c.flushPeriod)
 	return c
 }
 
-func (c *langfuseImpl) StartSendingEvents(ctx context.Context, period time.Duration) error {
-	if c.isShutdown.Load() {
-		return AlreadyShutdownErr
-	}
-	if c.isSendingEventsStarted.CompareAndSwap(false, true) {
-		ctx, cancel := context.WithCancel(ctx)
-		c.cancelFunc = cancel
-		go c.eventBuffer.Start(ctx, period)
-		return nil
-	} else {
-		return AlreadyStartedErr
-	}
+func (c *langfuseImpl) startSendingEvents(ctx context.Context, period time.Duration) {
+	ctx, cancel := context.WithCancel(ctx)
+	c.cancelFunc = cancel
+	go c.eventBuffer.Start(ctx, period)
 }
 
 func (c *langfuseImpl) Shutdown(ctx context.Context) error {
 	// Check if already shutdown
 	if c.isShutdown.CompareAndSwap(false, true) {
-		// Check if events were started
-		if !c.isSendingEventsStarted.Load() {
-			return nil
-		}
-
 		// Cancel the event buffer goroutine
 		c.cancelFunc()
 

langfuse_test.go

@@ -67,9 +67,8 @@ func (s *ClientTest) getLangfuseClientForTest(promptName, promptContent string)
 	"commitMessage" : null,
 	"resolutionGraph" : null
 }`, promptContent, promptName)
-	return golangfuse.NewWithHttpClient(
-		httpmock.NewMockClient(http.StatusOK, apiResponse),
-		"https://langfuse3.data.divar.cloud", "pk", "sk")
+	return golangfuse.New(s.ctx, "https://langfuse3.data.divar.cloud", "pk", "sk",
+		golangfuse.WithHTTPClient(httpmock.NewMockClient(http.StatusOK, apiResponse)))
 }
 
 func (s *ClientTest) TestShouldSetBasicAuth() {
@@ -91,19 +90,6 @@ func (s *ClientTest) TestShouldSetBasicAuth() {
 	)
 }
 
-func (s *ClientTest) TestStartSendingShouldReturnErrorIfAlreadyStarted() {
-	// Given
-	c := s.getClient()
-	err := c.StartSendingEvents(s.ctx, 1*time.Microsecond)
-	s.Require().NoError(err)
-
-	// When
-	err = c.StartSendingEvents(s.ctx, 1*time.Microsecond)
-
-	// Then
-	s.Require().ErrorContains(err, "already started")
-}
-
 func (s *ClientTest) TestShouldSendEvent() {
 	// Given
 	var sentRequestBody []byte
@@ -118,8 +104,6 @@ func (s *ClientTest) TestShouldSendEvent() {
 		sentRequestBody = body
 		return &http.Response{}, nil
 	})
-	err := c.StartSendingEvents(s.ctx, 1*time.Microsecond)
-	s.Require().NoError(err)
 
 	// When
 	c.Trace("input", "output")
@@ -135,7 +119,7 @@ func (s *ClientTest) TestShouldSendEvent() {
 		} `json:"batch"`
 	}
 	bodyObj := requestBody{}
-	err = json.Unmarshal(sentRequestBody, &bodyObj)
+	err := json.Unmarshal(sentRequestBody, &bodyObj)
 	s.Require().NoError(err)
 	s.Require().Len(bodyObj.Batch, 1)
 	s.Require().Equal("input", bodyObj.Batch[0].Body.Input)
@@ -145,15 +129,13 @@ func (s *ClientTest) TestShouldSendEvent() {
 func (s *ClientTest) TestShouldNotSendAnythingWhenNoEventIsReported() {
 	// Given
 	httpCallHappened := false
-	c := s.getClientWithMockedHttpTransport(func(req *http.Request) (*http.Response, error) {
+	s.getClientWithMockedHttpTransport(func(req *http.Request) (*http.Response, error) {
 		httpCallHappened = true
 		return &http.Response{}, nil
 	})
 
 	// When
-	err := c.StartSendingEvents(s.ctx, 1*time.Microsecond)
-	s.Require().NoError(err)
-	time.Sleep(1 * time.Millisecond)
+	time.Sleep(2 * time.Millisecond)
 
 	// Then
 	s.Require().False(httpCallHappened, "http call happened, unexpectedly")
@@ -173,8 +155,6 @@ func (s *ClientTest) TestShouldSendEventsInBatch() {
 		sentRequestBody = body
 		return &http.Response{}, nil
 	})
-	err := c.StartSendingEvents(s.ctx, 1*time.Microsecond)
-	s.Require().NoError(err)
 
 	// When
 	c.Trace("input", "output")
@@ -186,7 +166,7 @@ func (s *ClientTest) TestShouldSendEventsInBatch() {
 		Batch []struct{} `json:"batch"`
 	}
 	bodyObj := requestBody{}
-	err = json.Unmarshal(sentRequestBody, &bodyObj)
+	err := json.Unmarshal(sentRequestBody, &bodyObj)
 	s.Require().NoError(err)
 	s.Require().Len(bodyObj.Batch, 2)
 }
@@ -200,21 +180,24 @@ func (s *ClientTest) getClient() golangfuse.Langfuse {
 	})
 }
 
-func (s *ClientTest) getClientWithMockedHttpTransport(transport httpmock.RoundTripFunc) golangfuse.Langfuse {
-	return golangfuse.NewWithHttpClient(
-		&http.Client{Transport: transport},
+func (s *ClientTest) getClientWithMockedHttpTransport(
+	transport httpmock.RoundTripFunc,
+	opts ...golangfuse.Option,
+) golangfuse.Langfuse {
+	opts = append(opts, golangfuse.WithHTTPClient(&http.Client{Transport: transport}))
+	return golangfuse.New(
+		s.ctx,
 		"https://test.com",
 		"test-pk",
 		"test-sk",
+		opts...,
 	)
 }
 
 func (s *ClientTest) TestShutdownShouldReturnErrorWhenAlreadyShutdown() {
 	// Given
 	c := s.getClient()
-	err := c.StartSendingEvents(s.ctx, 1*time.Microsecond)
-	s.Require().NoError(err)
-	err = c.Shutdown(s.ctx)
+	err := c.Shutdown(s.ctx)
 	s.Require().NoError(err)
 
 	// When
@@ -224,14 +207,12 @@ func (s *ClientTest) TestShutdownShouldReturnErrorWhenAlreadyShutdown() {
 	s.Require().ErrorContains(err, "already shutdown")
 }
 
-func (s *ClientTest) TestShutdownShouldSucceedAfterStart() {
+func (s *ClientTest) TestShutdownShouldSucceedInHappyPath() {
 	// Given
 	c := s.getClient()
-	err := c.StartSendingEvents(s.ctx, 1*time.Microsecond)
-	s.Require().NoError(err)
 
 	// When
-	err = c.Shutdown(s.ctx)
+	err := c.Shutdown(s.ctx)
 
 	// Then
 	s.Require().NoError(err)
@@ -253,13 +234,13 @@ func (s *ClientTest) TestShutdownShouldFlushPendingEvents() {
 			StatusCode: http.StatusMultiStatus,
 			Body:       io.NopCloser(strings.NewReader("{}")),
 		}, nil
-	})
-	err := c.StartSendingEvents(s.ctx, 1*time.Hour) // Long period to prevent automatic Flush
-	s.Require().NoError(err)
+	},
+		golangfuse.WithFlushPeriod(1*time.Hour), // Long period to prevent automatic Flush
+	)
 
 	// When
 	c.Trace("input", "output")
-	err = c.Shutdown(s.ctx)
+	err := c.Shutdown(s.ctx)
 	s.Require().NoError(err)
 	wg.Wait()
 
@@ -280,21 +261,6 @@ func (s *ClientTest) TestShutdownShouldFlushPendingEvents() {
 	s.Require().Equal("output", bodyObj.Batch[0].Body.Output)
 }
 
-func (s *ClientTest) TestStartSendingEventsShouldReturnErrorAfterShutdown() {
-	// Given
-	c := s.getClient()
-	err := c.StartSendingEvents(s.ctx, 1*time.Microsecond)
-	s.Require().NoError(err)
-	err = c.Shutdown(s.ctx)
-	s.Require().NoError(err)
-
-	// When
-	err = c.StartSendingEvents(s.ctx, 1*time.Microsecond)
-
-	// Then
-	s.Require().ErrorContains(err, "already shutdown")
-}
-
 func TestLangfuseClient(t *testing.T) {
 	suite.Run(t, new(ClientTest))
 }

options.go

@@ -0,0 +1,31 @@
+package golangfuse
+
+import (
+	"net/http"
+	"time"
+
+	"resty.dev/v3"
+)
+
+type Option func(l *langfuseImpl)
+
+func WithDefaultPromptLabel(label string) Option {
+	return func(l *langfuseImpl) {
+		l.promptLabel = label
+	}
+}
+
+func WithHTTPClient(httpClient *http.Client) Option {
+	return func(l *langfuseImpl) {
+		l.restClient = resty.NewWithClient(httpClient)
+	}
+}
+
+func WithFlushPeriod(period time.Duration) Option {
+	return func(l *langfuseImpl) {
+		if period <= 0 {
+			period = 1 * time.Second // Default to 1 second if invalid period is provided
+		}
+		l.flushPeriod = period
+	}
+}