Moved RPC related types to a dedicated package (cadence-workflow#4505)

Moved RPC related types to a dedicated package

cmd/server/cadence/server.go

@@ -41,6 +41,7 @@ import (
 	"github.com/uber/cadence/common/log/tag"
 	"github.com/uber/cadence/common/messaging/kafka"
 	"github.com/uber/cadence/common/metrics"
+	"github.com/uber/cadence/common/rpc"
 	"github.com/uber/cadence/common/service"
 	"github.com/uber/cadence/service/frontend"
 	"github.com/uber/cadence/service/history"
@@ -150,7 +151,7 @@ func (s *server) startService() common.Daemon {
 
 	svcCfg := s.cfg.Services[s.name]
 	params.MetricScope = svcCfg.Metrics.NewScope(params.Logger, params.Name)
-	params.RPCFactory = svcCfg.RPC.NewFactory(params.Name, params.Logger, s.cfg.NewGRPCPorts())
+	params.RPCFactory = rpc.NewFactory(params.Name, svcCfg.RPC, params.Logger, rpc.NewGRPCPorts(s.cfg))
 	params.MembershipFactory, err = s.cfg.Ringpop.NewFactory(
 		params.RPCFactory.GetDispatcher(),
 		params.Name,

common/persistence/sql/sqldriver/interface.go

@@ -55,4 +55,4 @@ type (
 		GetContext(ctx context.Context, dbShardID int, dest interface{}, query string, args ...interface{}) error
 		SelectContext(ctx context.Context, dbShardID int, dest interface{}, query string, args ...interface{}) error
 	}
-)
+)

common/persistence/sql/sqldriver/singleton.go

@@ -41,8 +41,8 @@ type (
 // But this singleton doesn't have sharding so omitting it.
 func NewSingletonSQLDriver(xdb *sqlx.DB, xtx *sqlx.Tx, _ int) Driver {
 	driver := &singleton{
-		db:   xdb,
-		tx:   xtx,
+		db: xdb,
+		tx: xtx,
 	}
 	if xtx != nil {
 		driver.useTx = true
@@ -110,4 +110,4 @@ func (s *singleton) Commit() error {
 
 func (s *singleton) Rollback() error {
 	return s.tx.Rollback()
-}
+}

common/persistence/sql/sqlplugin/dbSharding.go

@@ -37,7 +37,7 @@ const (
 )
 
 // GetDBShardIDFromHistoryShardID maps  historyShardID to a DBShardID
-func GetDBShardIDFromHistoryShardID(historyShardID int, numDBShards int) int{
+func GetDBShardIDFromHistoryShardID(historyShardID int, numDBShards int) int {
 	return historyShardID % numDBShards
 }
 
@@ -57,4 +57,4 @@ func GetDBShardIDFromDomainID(domainID string, numDBShards int) int {
 func GetDBShardIDFromTreeID(treeID serialization.UUID, numDBShards int) int {
 	hash := farm.Hash32(treeID) % uint32(numDBShards)
 	return int(hash) % numDBShards
-}
+}

common/config/rpc.go → common/rpc/factory.go

@@ -18,29 +18,28 @@
 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 // THE SOFTWARE.
 
-package config
+package rpc
 
 import (
-	"errors"
 	"fmt"
 	"net"
-	"strings"
 	"sync"
 
 	"go.uber.org/yarpc"
 	"go.uber.org/yarpc/api/transport"
 	"go.uber.org/yarpc/transport/grpc"
 	"go.uber.org/yarpc/transport/tchannel"
 
+	"github.com/uber/cadence/common/config"
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/log/tag"
 )
 
-// RPCFactory is an implementation of service.RPCFactory interface
-type RPCFactory struct {
-	config      *RPC
+// Factory is an implementation of service.RPCFactory interface
+type Factory struct {
+	config      config.RPC
 	serviceName string
-	ch          *tchannel.ChannelTransport
+	tchannel    *tchannel.ChannelTransport
 	grpc        *grpc.Transport
 	logger      log.Logger
 	grpcPorts   GRPCPorts
@@ -49,19 +48,14 @@ type RPCFactory struct {
 	dispatcher *yarpc.Dispatcher
 }
 
-// NewFactory builds a new RPCFactory
+// NewFactory builds a new rpc.Factory
 // conforming to the underlying configuration
-func (cfg *RPC) NewFactory(sName string, logger log.Logger, grpcPorts GRPCPorts) *RPCFactory {
-	return newRPCFactory(cfg, sName, logger, grpcPorts)
-}
-
-func newRPCFactory(cfg *RPC, sName string, logger log.Logger, grpcPorts GRPCPorts) *RPCFactory {
-	factory := &RPCFactory{config: cfg, serviceName: sName, logger: logger, grpcPorts: grpcPorts}
-	return factory
+func NewFactory(service string, cfg config.RPC, logger log.Logger, grpcPorts GRPCPorts) *Factory {
+	return &Factory{config: cfg, serviceName: service, logger: logger, grpcPorts: grpcPorts}
 }
 
 // GetDispatcher return a cached dispatcher
-func (d *RPCFactory) GetDispatcher() *yarpc.Dispatcher {
+func (d *Factory) GetDispatcher() *yarpc.Dispatcher {
 	d.Lock()
 	defer d.Unlock()
 
@@ -74,19 +68,19 @@ func (d *RPCFactory) GetDispatcher() *yarpc.Dispatcher {
 }
 
 // createInboundDispatcher creates a dispatcher for inbound
-func (d *RPCFactory) createInboundDispatcher() *yarpc.Dispatcher {
+func (d *Factory) createInboundDispatcher() *yarpc.Dispatcher {
 	// Setup dispatcher for onebox
 	var err error
 	inbounds := yarpc.Inbounds{}
 
 	hostAddress := fmt.Sprintf("%v:%v", d.getListenIP(), d.config.Port)
-	d.ch, err = tchannel.NewChannelTransport(
+	d.tchannel, err = tchannel.NewChannelTransport(
 		tchannel.ServiceName(d.serviceName),
 		tchannel.ListenAddr(hostAddress))
 	if err != nil {
 		d.logger.Fatal("Failed to create transport channel", tag.Error(err))
 	}
-	inbounds = append(inbounds, d.ch.NewInbound())
+	inbounds = append(inbounds, d.tchannel.NewInbound())
 	d.logger.Info("Listening for TChannel requests", tag.Address(hostAddress))
 
 	var options []grpc.TransportOption
@@ -113,16 +107,16 @@ func (d *RPCFactory) createInboundDispatcher() *yarpc.Dispatcher {
 }
 
 // CreateDispatcherForOutbound creates a dispatcher for outbound connection
-func (d *RPCFactory) CreateDispatcherForOutbound(
+func (d *Factory) CreateDispatcherForOutbound(
 	callerName string,
 	serviceName string,
 	hostName string,
 ) (*yarpc.Dispatcher, error) {
-	return d.createOutboundDispatcher(callerName, serviceName, hostName, d.ch.NewSingleOutbound(hostName))
+	return d.createOutboundDispatcher(callerName, serviceName, hostName, d.tchannel.NewSingleOutbound(hostName))
 }
 
 // CreateGRPCDispatcherForOutbound creates a dispatcher for GRPC outbound connection
-func (d *RPCFactory) CreateGRPCDispatcherForOutbound(
+func (d *Factory) CreateGRPCDispatcherForOutbound(
 	callerName string,
 	serviceName string,
 	hostName string,
@@ -131,11 +125,11 @@ func (d *RPCFactory) CreateGRPCDispatcherForOutbound(
 }
 
 // ReplaceGRPCPort replaces port in the address to grpc for a given service
-func (d *RPCFactory) ReplaceGRPCPort(serviceName, hostAddress string) (string, error) {
+func (d *Factory) ReplaceGRPCPort(serviceName, hostAddress string) (string, error) {
 	return d.grpcPorts.GetGRPCAddress(serviceName, hostAddress)
 }
 
-func (d *RPCFactory) createOutboundDispatcher(
+func (d *Factory) createOutboundDispatcher(
 	callerName string,
 	serviceName string,
 	hostName string,
@@ -157,7 +151,7 @@ func (d *RPCFactory) createOutboundDispatcher(
 	return dispatcher, nil
 }
 
-func (d *RPCFactory) getListenIP() net.IP {
+func (d *Factory) getListenIP() net.IP {
 	if d.config.BindOnLocalHost && len(d.config.BindOnIP) > 0 {
 		d.logger.Fatal("ListenIP failed, bindOnLocalHost and bindOnIP are mutually exclusive")
 	}
@@ -179,30 +173,3 @@ func (d *RPCFactory) getListenIP() net.IP {
 	}
 	return ip
 }
-
-type GRPCPorts map[string]int
-
-func (c *Config) NewGRPCPorts() GRPCPorts {
-	grpcPorts := map[string]int{}
-	for service, config := range c.Services {
-		grpcPorts["cadence-"+service] = config.RPC.GRPCPort
-	}
-	return grpcPorts
-}
-
-func (p GRPCPorts) GetGRPCAddress(service, hostAddress string) (string, error) {
-	port, ok := p[service]
-	if !ok {
-		return hostAddress, errors.New("unknown service: " + service)
-	}
-	if port == 0 {
-		return hostAddress, errors.New("GRPC port not configured for service: " + service)
-	}
-
-	// Drop port if provided
-	if index := strings.Index(hostAddress, ":"); index > 0 {
-		hostAddress = hostAddress[:index]
-	}
-
-	return fmt.Sprintf("%s:%d", hostAddress, port), nil
-}

common/rpc/grpc.go

@@ -0,0 +1,56 @@
+// Copyright (c) 2021 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package rpc
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/uber/cadence/common/config"
+)
+
+type GRPCPorts map[string]int
+
+func NewGRPCPorts(c *config.Config) GRPCPorts {
+	grpcPorts := map[string]int{}
+	for service, config := range c.Services {
+		grpcPorts["cadence-"+service] = config.RPC.GRPCPort
+	}
+	return grpcPorts
+}
+
+func (p GRPCPorts) GetGRPCAddress(service, hostAddress string) (string, error) {
+	port, ok := p[service]
+	if !ok {
+		return hostAddress, errors.New("unknown service: " + service)
+	}
+	if port == 0 {
+		return hostAddress, errors.New("GRPC port not configured for service: " + service)
+	}
+
+	// Drop port if provided
+	if index := strings.Index(hostAddress, ":"); index > 0 {
+		hostAddress = hostAddress[:index]
+	}
+
+	return fmt.Sprintf("%s:%d", hostAddress, port), nil
+}

common/config/rpc_test.go → common/rpc/grpc_test.go

@@ -18,24 +18,25 @@
 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 // THE SOFTWARE.
 
-package config
+package rpc
 
 import (
 	"testing"
 
 	"github.com/uber/cadence/common"
+	"github.com/uber/cadence/common/config"
 
 	"github.com/stretchr/testify/assert"
 )
 
 func TestGRPCPorts(t *testing.T) {
-	config := Config{
-		Services: map[string]Service{
-			"frontend": {RPC: RPC{GRPCPort: 9999}},
-			"history":  {RPC: RPC{}},
+	config := config.Config{
+		Services: map[string]config.Service{
+			"frontend": {RPC: config.RPC{GRPCPort: 9999}},
+			"history":  {RPC: config.RPC{}},
 		},
 	}
-	grpcPorts := config.NewGRPCPorts()
+	grpcPorts := NewGRPCPorts(&config)
 
 	_, err := grpcPorts.GetGRPCAddress("some-service", "1.2.3.4")
 	assert.EqualError(t, err, "unknown service: some-service")

common/config/localip.go → common/rpc/localip.go

@@ -18,7 +18,7 @@
 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 // THE SOFTWARE.
 
-package config
+package rpc
 
 // ** This code is copied from tchannel, we would like to not take dependency on tchannel code **
 

common/config/localip_test.go → common/rpc/localip_test.go

@@ -18,7 +18,7 @@
 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 // THE SOFTWARE.
 
-package config
+package rpc
 
 // ** This code is copied from tchannel, we would like to not take dependency on tchannel code **