Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: use just one grpc connection for topics #360

Merged
merged 2 commits into from
Oct 20, 2023
Merged

fix: use just one grpc connection for topics #360

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9cb04ff
fix: use just one grpc connection for topics
anitarua Oct 20, 2023
1d6fd8f
publish should use next topic manager in rotation instead of reserved…
anitarua Oct 20, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions internal/grpcmanagers/topic_manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ func NewStreamTopicGrpcManager(request *models.TopicStreamGrpcManagerRequest) (*
endpoint,
grpc.WithTransportCredentials(credentials.NewTLS(config)),
grpc.WithChainStreamInterceptor(interceptor.AddStreamHeaderInterceptor(authToken)),
grpc.WithChainUnaryInterceptor(interceptor.AddHeadersInterceptor(authToken)),
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

adds authorization header to any unary requests made using this connection

)

if err != nil {
Expand Down
22 changes: 5 additions & 17 deletions momento/pubsub_client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,15 +9,12 @@ import (
"github.com/momentohq/client-sdk-go/internal/models"
"github.com/momentohq/client-sdk-go/internal/momentoerrors"
pb "github.com/momentohq/client-sdk-go/internal/protos"
"github.com/momentohq/client-sdk-go/internal/retry"

"google.golang.org/grpc"
)

type pubSubClient struct {
streamTopicManagers []*grpcmanagers.TopicGrpcManager
unaryDataManager *grpcmanagers.DataGrpcManager
unaryGrpcClient pb.PubsubClient
endpoint string
}

Expand All @@ -29,7 +26,8 @@ func newPubSubClient(request *models.PubSubClientRequest) (*pubSubClient, moment
if numSubscriptions > 0 {
// a single channel can support 100 streams, so we need to create enough
// channels to handle the maximum number of subscriptions
numChannels = uint32(math.Ceil(numSubscriptions / 100.0))
// plus one for the publishing channel
numChannels = uint32(math.Ceil((numSubscriptions + 1) / 100.0))
} else {
numChannels = 1
}
Expand All @@ -45,18 +43,8 @@ func newPubSubClient(request *models.PubSubClientRequest) (*pubSubClient, moment
streamTopicManagers = append(streamTopicManagers, streamTopicManager)
}

unaryDataManager, err := grpcmanagers.NewUnaryDataGrpcManager(&models.DataGrpcManagerRequest{
CredentialProvider: request.CredentialProvider,
RetryStrategy: retry.NewNeverRetryStrategy(),
})
if err != nil {
return nil, err
}

return &pubSubClient{
streamTopicManagers: streamTopicManagers,
unaryDataManager: unaryDataManager,
unaryGrpcClient: pb.NewPubsubClient(unaryDataManager.Conn),
endpoint: request.CredentialProvider.GetCacheEndpoint(),
}, nil
}
Expand All @@ -78,9 +66,10 @@ func (client *pubSubClient) topicSubscribe(ctx context.Context, request *TopicSu
}

func (client *pubSubClient) topicPublish(ctx context.Context, request *TopicPublishRequest) error {
topicManager := client.getNextStreamTopicManager()
switch value := request.Value.(type) {
case String:
_, err := client.unaryGrpcClient.Publish(ctx, &pb.XPublishRequest{
_, err := topicManager.StreamClient.Publish(ctx, &pb.XPublishRequest{
CacheName: request.CacheName,
Topic: request.TopicName,
Value: &pb.XTopicValue{
Expand All @@ -91,7 +80,7 @@ func (client *pubSubClient) topicPublish(ctx context.Context, request *TopicPubl
})
return err
case Bytes:
_, err := client.unaryGrpcClient.Publish(ctx, &pb.XPublishRequest{
_, err := topicManager.StreamClient.Publish(ctx, &pb.XPublishRequest{
CacheName: request.CacheName,
Topic: request.TopicName,
Value: &pb.XTopicValue{
Expand All @@ -113,5 +102,4 @@ func (client *pubSubClient) close() {
for clientIndex := range client.streamTopicManagers {
defer client.streamTopicManagers[clientIndex].Close()
}
defer client.unaryDataManager.Close()
}