some cleanups in proxy code (#944)

Author: FZambia

internal/proxy/connect_grpc.go

@@ -55,5 +55,5 @@ func (p *GRPCConnectProxy) UseBase64() bool {
 func (p *GRPCConnectProxy) ProxyConnect(ctx context.Context, req *proxyproto.ConnectRequest) (*proxyproto.ConnectResponse, error) {
 	ctx, cancel := context.WithTimeout(ctx, p.config.Timeout.ToDuration())
 	defer cancel()
-	return p.client.Connect(grpcRequestContext(ctx, p.config), req, grpc.ForceCodec(grpcCodec))
+	return p.client.Connect(grpcRequestContext(ctx, p.config), req)
 }

internal/proxy/grpc.go

@@ -9,7 +9,6 @@ import (
 
 	"github.com/centrifugal/centrifugo/v6/internal/clientcontext"
 	"github.com/centrifugal/centrifugo/v6/internal/middleware"
-	"github.com/centrifugal/centrifugo/v6/internal/proxyproto"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
@@ -18,8 +17,6 @@ import (
 	"google.golang.org/grpc/metadata"
 )
 
-var grpcCodec = proxyproto.Codec{}
-
 type rpcCredentials struct {
 	key   string
 	value string

internal/proxy/publish_grpc.go

@@ -41,7 +41,7 @@ func NewGRPCPublishProxy(name string, p Config) (*GRPCPublishProxy, error) {
 func (p *GRPCPublishProxy) ProxyPublish(ctx context.Context, req *proxyproto.PublishRequest) (*proxyproto.PublishResponse, error) {
 	ctx, cancel := context.WithTimeout(ctx, p.config.Timeout.ToDuration())
 	defer cancel()
-	return p.client.Publish(grpcRequestContext(ctx, p.config), req, grpc.ForceCodec(grpcCodec))
+	return p.client.Publish(grpcRequestContext(ctx, p.config), req)
 }
 
 // Protocol ...

internal/proxy/refresh_grpc.go

@@ -41,7 +41,7 @@ func NewGRPCRefreshProxy(name string, p Config) (*GRPCRefreshProxy, error) {
 func (p *GRPCRefreshProxy) ProxyRefresh(ctx context.Context, req *proxyproto.RefreshRequest) (*proxyproto.RefreshResponse, error) {
 	ctx, cancel := context.WithTimeout(ctx, p.config.Timeout.ToDuration())
 	defer cancel()
-	return p.client.Refresh(grpcRequestContext(ctx, p.config), req, grpc.ForceCodec(grpcCodec))
+	return p.client.Refresh(grpcRequestContext(ctx, p.config), req)
 }
 
 // Name ...

internal/proxy/refresh_http.go

@@ -7,13 +7,6 @@ import (
 	"github.com/centrifugal/centrifugo/v6/internal/proxyproto"
 )
 
-// RefreshRequestHTTP ...
-type RefreshRequestHTTP struct {
-	baseRequestHTTP
-
-	UserID string `json:"user"`
-}
-
 // HTTPRefreshProxy ...
 type HTTPRefreshProxy struct {
 	config     Config

internal/proxy/rpc_grpc.go

@@ -41,7 +41,7 @@ func NewGRPCRPCProxy(name string, p Config) (*GRPCRPCProxy, error) {
 func (p *GRPCRPCProxy) ProxyRPC(ctx context.Context, req *proxyproto.RPCRequest) (*proxyproto.RPCResponse, error) {
 	ctx, cancel := context.WithTimeout(ctx, p.config.Timeout.ToDuration())
 	defer cancel()
-	return p.client.RPC(grpcRequestContext(ctx, p.config), req, grpc.ForceCodec(grpcCodec))
+	return p.client.RPC(grpcRequestContext(ctx, p.config), req)
 }
 
 // Protocol ...

internal/proxy/sub_refresh_grpc.go

@@ -41,7 +41,7 @@ func NewGRPCSubRefreshProxy(name string, p Config) (*GRPCSubRefreshProxy, error)
 func (p *GRPCSubRefreshProxy) ProxySubRefresh(ctx context.Context, req *proxyproto.SubRefreshRequest) (*proxyproto.SubRefreshResponse, error) {
 	ctx, cancel := context.WithTimeout(ctx, p.config.Timeout.ToDuration())
 	defer cancel()
-	return p.client.SubRefresh(grpcRequestContext(ctx, p.config), req, grpc.ForceCodec(grpcCodec))
+	return p.client.SubRefresh(grpcRequestContext(ctx, p.config), req)
 }
 
 // Protocol ...

internal/proxy/sub_refresh_http.go

@@ -7,14 +7,6 @@ import (
 	"github.com/centrifugal/centrifugo/v6/internal/proxyproto"
 )
 
-// SubRefreshRequestHTTP ...
-type SubRefreshRequestHTTP struct {
-	baseRequestHTTP
-
-	UserID  string `json:"user"`
-	Channel string `json:"channel"`
-}
-
 // HTTPSubRefreshProxy ...
 type HTTPSubRefreshProxy struct {
 	config     Config

internal/proxy/subscribe_grpc.go

@@ -41,7 +41,7 @@ func NewGRPCSubscribeProxy(name string, p Config) (*GRPCSubscribeProxy, error) {
 func (p *GRPCSubscribeProxy) ProxySubscribe(ctx context.Context, req *proxyproto.SubscribeRequest) (*proxyproto.SubscribeResponse, error) {
 	ctx, cancel := context.WithTimeout(ctx, p.config.Timeout.ToDuration())
 	defer cancel()
-	return p.client.Subscribe(grpcRequestContext(ctx, p.config), req, grpc.ForceCodec(grpcCodec))
+	return p.client.Subscribe(grpcRequestContext(ctx, p.config), req)
 }
 
 // Protocol ...

internal/proxy/subscribe_stream_grpc.go

@@ -36,10 +36,10 @@ func NewSubscribeStreamProxy(name string, p Config) (*SubscribeStreamProxy, erro
 
 // SubscribeUnidirectional ...
 func (p *SubscribeStreamProxy) SubscribeUnidirectional(ctx context.Context, req *proxyproto.SubscribeRequest) (proxyproto.CentrifugoProxy_SubscribeUnidirectionalClient, error) {
-	return p.client.SubscribeUnidirectional(grpcRequestContext(ctx, p.config), req, grpc.ForceCodec(grpcCodec))
+	return p.client.SubscribeUnidirectional(grpcRequestContext(ctx, p.config), req)
 }
 
 // SubscribeBidirectional ...
 func (p *SubscribeStreamProxy) SubscribeBidirectional(ctx context.Context) (proxyproto.CentrifugoProxy_SubscribeBidirectionalClient, error) {
-	return p.client.SubscribeBidirectional(grpcRequestContext(ctx, p.config), grpc.ForceCodec(grpcCodec))
+	return p.client.SubscribeBidirectional(grpcRequestContext(ctx, p.config))
 }