Skip to content

Commit

Permalink
Refactor the structure of multiplexing listener implementation
Browse files Browse the repository at this point in the history
  • Loading branch information
Shibo Wang committed Jun 2, 2021
1 parent 1292bf4 commit 36be9a3
Show file tree
Hide file tree
Showing 4 changed files with 107 additions and 4 deletions.
42 changes: 40 additions & 2 deletions config/configgrpc/configgrpc.go
Original file line number Diff line number Diff line change
Expand Up @@ -235,9 +235,47 @@ func validateBalancerName(balancerName string) bool {
return false
}

// ToListener returns the net.Listener constructed from the settings.
// muxListeners is a map of Endpoint to multiplexing listener
var muxListeners = make(map[string]net.Listener)

// ToListener creates a new net.Listener when it is called at the first time. After that,
// whenever the same Endpoint is passed in, that listener will be returned.
func (gss *GRPCServerSettings) ToListener() (net.Listener, error) {
return gss.NetAddr.Listen()
var addr net.Addr
var err error

// Get the network address according to the protocol type and the given endpoint.
switch gss.NetAddr.Transport {
case "tcp", "tcp4", "tcp6":
addr, err = net.ResolveTCPAddr(gss.NetAddr.Transport, gss.NetAddr.Endpoint)
case "udp", "udp4", "udp6":
addr, err = net.ResolveUDPAddr(gss.NetAddr.Transport, gss.NetAddr.Endpoint)
case "ip", "ip4", "ip6":
addr, err = net.ResolveIPAddr(gss.NetAddr.Transport, gss.NetAddr.Endpoint)
case "unix", "unixgram", "unixpacket":
addr, err = net.ResolveUnixAddr(gss.NetAddr.Transport, gss.NetAddr.Endpoint)
default:
return nil, fmt.Errorf("unknown network protocol")
}
if err != nil {
return nil, err
}

// If the Endpoint is in use (i.e. it corresponds to a listener), use the existing
// listener. Otherwise, get a new listener.
if muxLs, ok := muxListeners[addr.String()]; ok {
return muxLs, nil
}

listener, err := gss.NetAddr.Listen()
if err != nil {
return nil, err
}

// Set the multiplexing listener.
muxListeners[addr.String()] = listener

return listener, nil
}

// ToServerOption maps configgrpc.GRPCServerSettings to a slice of server options for gRPC
Expand Down
22 changes: 22 additions & 0 deletions config/configgrpc/configgrpc_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -536,3 +536,25 @@ func tempSocketName(t *testing.T) string {
require.NoError(t, os.Remove(socket))
return socket
}

func TestGRPCPortReuse(t *testing.T) {
endpoint := "localhost: 2873"
t.Run("", func(t *testing.T) {
settings1 := createGRPCServerSettings(endpoint)
muxListener1, err := settings1.ToListener()
assert.NoError(t, err)
settings2 := createGRPCServerSettings(endpoint)
muxListener2, err := settings2.ToListener()
assert.NoError(t, err)
assert.Equal(t, muxListener1, muxListener2)
})
}

func createGRPCServerSettings(endpoint string) GRPCServerSettings {
return GRPCServerSettings{
NetAddr: confignet.NetAddr{
Endpoint: endpoint,
Transport: "tcp",
},
}
}
28 changes: 26 additions & 2 deletions config/confighttp/confighttp.go
Original file line number Diff line number Diff line change
Expand Up @@ -125,13 +125,33 @@ type HTTPServerSettings struct {
CorsHeaders []string `mapstructure:"cors_allowed_headers"`
}

// ToListener creates a net.Listener.
// muxListeners is a map of Endpoint to multiplexing listener
var muxListeners = make(map[string]net.Listener)

// ToListener creates a new net.Listener when it is called at the first time. After that,
// whenever the same Endpoint is passed in, that listener will be returned.
func (hss *HTTPServerSettings) ToListener() (net.Listener, error) {
listener, err := net.Listen("tcp", hss.Endpoint)
var listener net.Listener
var addr net.Addr
var err error

// Get the tcp address according to the given endpoint.
addr, err = net.ResolveTCPAddr("tcp", hss.Endpoint)
if err != nil {
return nil, err
}

// If the Endpoint is in use (i.e. it corresponds to a listener), use the existing
// listener. Otherwise, get a new listener.
if muxLs, ok := muxListeners[addr.String()]; ok {
listener = muxLs
} else {
listener, err = net.Listen("tcp", hss.Endpoint)
if err != nil {
return nil, err
}
}

if hss.TLSSetting != nil {
var tlsCfg *tls.Config
tlsCfg, err = hss.TLSSetting.LoadTLSConfig()
Expand All @@ -140,6 +160,10 @@ func (hss *HTTPServerSettings) ToListener() (net.Listener, error) {
}
listener = tls.NewListener(listener, tlsCfg)
}

// Set the multiplexing listener.
muxListeners[addr.String()] = listener

return listener, nil
}

Expand Down
19 changes: 19 additions & 0 deletions config/confighttp/confighttp_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -482,3 +482,22 @@ func TestHttpHeaders(t *testing.T) {
})
}
}

func TestHTTPPortReuse(t *testing.T) {
endpoint := "localhost: 6489"
t.Run("", func(t *testing.T) {
settings1 := createHTTPServerSettings(endpoint)
muxListener1, err := settings1.ToListener()
assert.NoError(t, err)
settings2 := createHTTPServerSettings(endpoint)
muxListener2, err := settings2.ToListener()
assert.NoError(t, err)
assert.Equal(t, muxListener1, muxListener2)
})
}

func createHTTPServerSettings(endpoint string) HTTPServerSettings {
return HTTPServerSettings{
Endpoint: endpoint,
}
}

0 comments on commit 36be9a3

Please sign in to comment.