chore: add check for context cancel or timeout

.github/workflows/build.yml

@@ -61,6 +61,6 @@ jobs:
       - name: Run test
         run: |
           go vet ./...
-          go test -v ./momento
+          go test -v ./momento ./incubating
 
 

.github/workflows/manual_release.yml

@@ -67,7 +67,7 @@ jobs:
       - name: Run test
         run: |
           go vet ./...
-          go test -v ./momento
+          go test -v ./momento ./incubating
 
       - name: Publish package
         run: |

.github/workflows/on-push-to-main.yml

@@ -47,7 +47,7 @@ jobs:
       - name: Run test
         run: |
           go vet ./...
-          go test -v ./momento
+          go test -v ./momento ./incubating
 
       - name: Generate README
         uses: momentohq/standards-and-practices/github-actions/generate-and-commit-oss-readme@gh-actions-v1

examples/pubsub-example/go.mod

@@ -0,0 +1,19 @@
+module github.com/momentohq/go-pubsub-example
+
+go 1.19
+
+require (
+	github.com/google/uuid v1.3.0
+	github.com/momentohq/client-sdk-go v0.6.0
+)
+
+require (
+	github.com/golang-jwt/jwt/v4 v4.3.0 // indirect
+	github.com/golang/protobuf v1.5.2 // indirect
+	golang.org/x/net v0.4.0 // indirect
+	golang.org/x/sys v0.3.0 // indirect
+	golang.org/x/text v0.5.0 // indirect
+	google.golang.org/genproto v0.0.0-20221118155620-16455021b5e6 // indirect
+	google.golang.org/grpc v1.52.0 // indirect
+	google.golang.org/protobuf v1.28.1 // indirect
+)

examples/pubsub-example/go.sum

@@ -0,0 +1,16 @@
+github.com/golang-jwt/jwt/v4 v4.3.0/go.mod h1:/xlHOz8bRuivTWchD4jCa+NbatV+wEUSzwAxVc6locg=
+github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
+github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
+github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/momentohq/client-sdk-go v0.6.0 h1:pA8c5yZ7zeCidNqmXhh2Dt2IvaE1BSMVIh3fwUsyfK4=
+github.com/momentohq/client-sdk-go v0.6.0/go.mod h1:9llwnDsZ87XAACS5slQKqCcYhfLDSHxAMcIbQWDbdJw=
+golang.org/x/net v0.4.0/go.mod h1:MBQ8lrhLObU/6UmLb4fmbmk5OcyYmqtbGd/9yIeKjEE=
+golang.org/x/sys v0.3.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/text v0.5.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/genproto v0.0.0-20221118155620-16455021b5e6/go.mod h1:rZS5c/ZVYMaOGBfO68GWtjOw/eLaZM1X6iVtgjZ+EWg=
+google.golang.org/grpc v1.52.0/go.mod h1:pu6fVzoFb+NBYNAvQL08ic+lvB2IojljRYuun5vorUY=
+google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
+google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
+google.golang.org/protobuf v1.28.1/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=

examples/pubsub-example/main.go

@@ -0,0 +1,106 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/momentohq/client-sdk-go/auth"
+	"github.com/momentohq/client-sdk-go/config"
+	"github.com/momentohq/client-sdk-go/incubating"
+	"github.com/momentohq/client-sdk-go/momento"
+)
+
+const (
+	cacheName = "test-cache"
+	topicName = "test-topic"
+)
+
+func main() {
+	// Initialization
+	client := getClient()
+	ctx := context.Background()
+	cancelContext, cancelFunction := context.WithCancel(ctx)
+	// Or you can set a timeout after which the goroutine will be cancelled
+	//cancelContext, cancelFunction := context.WithTimeout(ctx, time.Second*10)
+	setupCache(client, ctx)
+
+	// Instantiate subscriber
+	sub, err := client.SubscribeTopic(ctx, &incubating.TopicSubscribeRequest{
+		CacheName: cacheName,
+		TopicName: topicName,
+	})
+	if err != nil {
+		panic(err)
+	}
+	// Receive and print messages in a goroutine
+	go func() { pollForMessages(sub, cancelContext) }()
+
+	// Publish messages and then shut down the subscriber goroutine
+	publishMessages(client, ctx)
+	cancelFunction()
+	// Prove that the goroutine is stopped by publishing more messages that
+	// won't be output to the console
+	publishMessages(client, ctx)
+}
+
+func getClient() incubating.ScsClient {
+	credProvider, err := auth.NewEnvMomentoTokenProvider("TEST_AUTH_TOKEN")
+	if err != nil {
+		panic(err)
+	}
+	client, err := incubating.NewScsClient(&momento.SimpleCacheClientProps{
+		Configuration:      config.LatestLaptopConfig(),
+		CredentialProvider: credProvider,
+	})
+	if err != nil {
+		panic(err)
+	}
+	return client
+}
+
+func setupCache(client momento.ScsClient, ctx context.Context) {
+	err := client.CreateCache(ctx, &momento.CreateCacheRequest{
+		CacheName: "test-cache",
+	})
+	if err != nil {
+		var momentoErr momento.MomentoError
+		if errors.As(err, &momentoErr) {
+			if momentoErr.Code() != momento.AlreadyExistsError {
+				panic(err)
+			}
+		}
+	}
+}
+
+func publishMessages(client incubating.ScsClient, ctx context.Context) {
+	for i := 0; i < 10; i++ {
+		fmt.Printf("publishing message %d\n", i)
+		err := client.PublishTopic(ctx, &incubating.TopicPublishRequest{
+			CacheName: cacheName,
+			TopicName: topicName,
+			Value: &incubating.TopicValueString{
+				Text: fmt.Sprintf("hello %d", i),
+			},
+		})
+		if err != nil {
+			panic(err)
+		}
+		time.Sleep(time.Second)
+	}
+}
+
+func pollForMessages(sub incubating.SubscriptionIFace, cancelContext context.Context) {
+	err := sub.Recv(cancelContext, func(ctx context.Context, m incubating.TopicValue) {
+		switch msg := m.(type) {
+		case *incubating.TopicValueString:
+			fmt.Printf("received message: '%s'\n", msg.Text)
+		case *incubating.TopicValueBytes:
+			fmt.Printf("received message: '%s'\n", msg.Bytes)
+		}
+	})
+	if err != nil {
+		panic(err)
+	}
+}

go.work

@@ -3,4 +3,5 @@ go 1.19
 use (
 	.
 	./examples
+	examples/pubsub-example
 )