mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-30 14:56:19 +00:00
Roey BermanandGitHub
c788728852
## What changed? Added a protoc plugin to generate client for internal CHASM requests to the history service. A CHASM library can now define its own set of gRPC services and register a handler for those. I had to modify the history client files to be generic, which broke the mock generation, so I had to hand code the tests there, which IMHO is actually nicer than the way they were with mocks. Also opted out of mock generation for CHASM services, there's just not enough value there. **NOTE**: Instead of creating three separate clients, I generated a single layered client, with a constructor that is DI friendly. There doesn't seem to be a good enough reason to break this out into separate clients or make client construction configurable (for now). **NOTE**: We can also get rid of the genrpcwrappers script eventually and use the protoc plugin approach for all services, I decided not to do that here though to reduce scope. For a preview of the generated file see: - https://github.com/bergundy/temporal/blob/3a89614a9097eca9061b5f347c01eda735534bbe/chasm/lib/activity/proto/v1/service.proto - https://github.com/bergundy/temporal/blob/3a89614a9097eca9061b5f347c01eda735534bbe/chasm/lib/activity/gen/activitypb/v1/service_client.pb.go ## Why? Keep all CHASM library functionality contained. ## How did you test it? - [ ] built
94 lines
2.1 KiB
Go
94 lines
2.1 KiB
Go
package history
|
|
|
|
import (
|
|
"sync"
|
|
|
|
"go.temporal.io/server/common/membership"
|
|
"google.golang.org/grpc"
|
|
)
|
|
|
|
type (
|
|
clientConnection[C any] struct {
|
|
grpcClient C
|
|
grpcConn *grpc.ClientConn
|
|
}
|
|
|
|
rpcAddress string
|
|
|
|
connectionPoolImpl[C any] struct {
|
|
mu struct {
|
|
sync.RWMutex
|
|
conns map[rpcAddress]clientConnection[C]
|
|
}
|
|
|
|
historyServiceResolver membership.ServiceResolver
|
|
rpcFactory RPCFactory
|
|
clientCtor func(grpc.ClientConnInterface) C
|
|
}
|
|
|
|
// RPCFactory is a subset of the [go.temporal.io/server/common/rpc.RPCFactory] interface to make testing easier.
|
|
RPCFactory interface {
|
|
CreateHistoryGRPCConnection(rpcAddress string) *grpc.ClientConn
|
|
}
|
|
|
|
connectionPool[C any] interface {
|
|
getOrCreateClientConn(addr rpcAddress) clientConnection[C]
|
|
getAllClientConns() []clientConnection[C]
|
|
resetConnectBackoff(clientConnection[C])
|
|
}
|
|
)
|
|
|
|
func NewConnectionPool[C any](
|
|
historyServiceResolver membership.ServiceResolver,
|
|
rpcFactory RPCFactory,
|
|
clientCtor func(grpc.ClientConnInterface) C,
|
|
) *connectionPoolImpl[C] {
|
|
c := &connectionPoolImpl[C]{
|
|
historyServiceResolver: historyServiceResolver,
|
|
rpcFactory: rpcFactory,
|
|
clientCtor: clientCtor,
|
|
}
|
|
c.mu.conns = make(map[rpcAddress]clientConnection[C])
|
|
return c
|
|
}
|
|
|
|
func (c *connectionPoolImpl[C]) getOrCreateClientConn(addr rpcAddress) clientConnection[C] {
|
|
c.mu.RLock()
|
|
cc, ok := c.mu.conns[addr]
|
|
c.mu.RUnlock()
|
|
if ok {
|
|
return cc
|
|
}
|
|
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
|
|
if cc, ok = c.mu.conns[addr]; ok {
|
|
return cc
|
|
}
|
|
grpcConn := c.rpcFactory.CreateHistoryGRPCConnection(string(addr))
|
|
cc = clientConnection[C]{
|
|
grpcClient: c.clientCtor(grpcConn),
|
|
grpcConn: grpcConn,
|
|
}
|
|
|
|
c.mu.conns[addr] = cc
|
|
return cc
|
|
}
|
|
|
|
func (c *connectionPoolImpl[C]) getAllClientConns() []clientConnection[C] {
|
|
hostInfos := c.historyServiceResolver.Members()
|
|
|
|
var clientConns []clientConnection[C]
|
|
for _, hostInfo := range hostInfos {
|
|
cc := c.getOrCreateClientConn(rpcAddress(hostInfo.GetAddress()))
|
|
clientConns = append(clientConns, cc)
|
|
}
|
|
|
|
return clientConns
|
|
}
|
|
|
|
func (c *connectionPoolImpl[C]) resetConnectBackoff(cc clientConnection[C]) {
|
|
cc.grpcConn.ResetConnectBackoff()
|
|
}
|