Tests

cluster/grpc_rpc_client.go

@@ -157,6 +157,7 @@ func (gs *GRPCClient) SendKick(userID string, serverType string, kick *protos.Ki
 	if gs.bindingStorage == nil {
 		return constants.ErrNoBindingStorageModule
 	}
+
 	svID, err = gs.bindingStorage.GetUserFrontendID(userID, serverType)
 	if err != nil {
 		return err

cluster/grpc_rpc_client_test.go

@@ -122,6 +122,66 @@ func TestBroadcastSessionBind(t *testing.T) {
 	}
 }
 
+func TestSendKick(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	mockBindingStorage := mocks.NewMockBindingStorage(ctrl)
+	mockPitayaClient := protosmocks.NewMockPitayaClient(ctrl)
+	tables := []struct {
+		name           string
+		userId         string
+		bindingStorage interfaces.BindingStorage
+		sv             *Server
+		err            error
+	}{
+		{"bindingstorage", "uid", mockBindingStorage, &Server{
+			Type:     "tp",
+			Frontend: true}, nil,
+		},
+		{"nobindingstorage", "uid", nil, &Server{
+			Type:     "tp",
+			Frontend: true,
+		}, constants.ErrNoBindingStorageModule},
+		{"nobindingstorage", "", mockBindingStorage, &Server{
+			Type:     "tp",
+			Frontend: true,
+		}, nil},
+	}
+
+	for _, table := range tables {
+		t.Run(table.name, func(t *testing.T) {
+			c := getConfig()
+			g, err := getRPCClient(c)
+			assert.NoError(t, err)
+
+			if table.bindingStorage != nil {
+				g.clientMap.Store(table.sv.ID, mockPitayaClient)
+				g.bindingStorage = table.bindingStorage
+				mockBindingStorage.EXPECT().GetUserFrontendID(table.userId, gomock.Any()).DoAndReturn(func(u, svType string) (string, error) {
+					assert.Equal(t, table.userId, u)
+					assert.Equal(t, table.sv.Type, svType)
+					return table.sv.ID, nil
+				})
+
+				mockPitayaClient.EXPECT().KickUser(gomock.Any(), gomock.Any()).Do(func(ctx context.Context, msg *protos.KickMsg) {
+					assert.Equal(t, table.userId, msg.UserId)
+				})
+			}
+
+			kick := &protos.KickMsg{
+				UserId: table.userId,
+			}
+
+			err = g.SendKick(table.userId, table.sv.Type, kick)
+			if table.err != nil {
+				assert.Equal(t, err, table.err)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}
+
 func TestSendPush(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()

cluster/nats_rpc_client_test.go

@@ -167,6 +167,42 @@ func TestNatsRPCClientBroadcastSessionBind(t *testing.T) {
 	subs.Unsubscribe()
 }
 
+func TestNatsRPCClientSendKick(t *testing.T) {
+	uid := "testuid"
+	s := helpers.GetTestNatsServer(t)
+	defer s.Shutdown()
+	cfg := viper.New()
+	cfg.Set("pitaya.cluster.rpc.client.nats.connect", fmt.Sprintf("nats://%s", s.Addr()))
+	config := getConfig(cfg)
+	sv := getServer()
+
+	rpcClient, _ := NewNatsRPCClient(config, sv, nil, nil)
+	err := rpcClient.Init()
+	assert.NoError(t, err)
+
+	kickChan := make(chan *nats.Msg)
+	subs, err := rpcClient.conn.ChanSubscribe(GetUserKickTopic(uid, sv.Type), kickChan)
+	assert.NoError(t, err)
+	time.Sleep(50 * time.Millisecond)
+
+	kick := &protos.KickMsg{
+		UserId: uid,
+	}
+
+	err = rpcClient.SendKick(uid, sv.Type, kick)
+	assert.NoError(t, err)
+
+	m := helpers.ShouldEventuallyReceive(t, kickChan).(*nats.Msg)
+
+	actual := &protos.KickMsg{}
+	err = proto.Unmarshal(m.Data, actual)
+	assert.NoError(t, err)
+
+	assert.Equal(t, kick.UserId, actual.UserId)
+	err = subs.Unsubscribe()
+	assert.NoError(t, err)
+}
+
 func TestNatsRPCClientSendPush(t *testing.T) {
 	uid := "testuid123"
 	s := helpers.GetTestNatsServer(t)

cluster/nats_rpc_server_test.go

@@ -91,6 +91,14 @@ func TestNatsRPCServerGetUserMessagesTopic(t *testing.T) {
 	assert.Equal(t, "pitaya/connector/user/1/push", GetUserMessagesTopic("1", "connector"))
 }
 
+func TestNatsRPCServerGetUserKickTopic(t *testing.T) {
+	t.Parallel()
+	assert.Equal(t, "pitaya/connector/user/0/kick", GetUserKickTopic("0", "connector"))
+	assert.Equal(t, "pitaya/game/user/1/kick", GetUserKickTopic("1", "game"))
+	assert.Equal(t, "pitaya/connector/user/10/kick", GetUserKickTopic("10", "connector"))
+	assert.Equal(t, "pitaya/game/user/11/kick", GetUserKickTopic("11", "game"))
+}
+
 func TestNatsRPCServerGetUnhandledRequestsChannel(t *testing.T) {
 	t.Parallel()
 	cfg := getConfig()
@@ -123,6 +131,7 @@ func TestNatsRPCServerOnSessionBind(t *testing.T) {
 	err = rpcServer.onSessionBind(context.Background(), sess)
 	assert.NoError(t, err)
 	assert.NotNil(t, sess.Subscription)
+	assert.NotNil(t, rpcServer.userKickCh)
 }
 
 func TestNatsRPCServerSubscribeToBindingsChannel(t *testing.T) {
@@ -143,6 +152,25 @@ func TestNatsRPCServerSubscribeToBindingsChannel(t *testing.T) {
 	assert.Equal(t, msg.Data, dt)
 }
 
+func TestNatsRPCServerSubscribeUserKickChannel(t *testing.T) {
+	t.Parallel()
+	cfg := getConfig()
+	sv := getServer()
+	rpcServer, _ := NewNatsRPCServer(cfg, sv, nil, nil)
+	s := helpers.GetTestNatsServer(t)
+	defer s.Shutdown()
+	conn, err := setupNatsConn(fmt.Sprintf("nats://%s", s.Addr()), nil)
+	assert.NoError(t, err)
+	rpcServer.conn = conn
+	err = rpcServer.subscribeToUserKickChannel("someuid", sv.Type)
+	assert.NoError(t, err)
+	dt := []byte("somedata")
+	err = conn.Publish(GetUserKickTopic("someuid", sv.Type), dt)
+	assert.NoError(t, err)
+	msg := helpers.ShouldEventuallyReceive(t, rpcServer.getUserKickChannel()).(*nats.Msg)
+	assert.Equal(t, msg.Data, dt)
+}
+
 func TestNatsRPCServerGetUserPushChannel(t *testing.T) {
 	t.Parallel()
 	cfg := getConfig()
@@ -152,6 +180,14 @@ func TestNatsRPCServerGetUserPushChannel(t *testing.T) {
 	assert.IsType(t, make(chan *protos.Push), n.getUserPushChannel())
 }
 
+func TestNatsRPCServerGetUserKickChannel(t *testing.T) {
+	t.Parallel()
+	cfg := getConfig()
+	sv := getServer()
+	n, _ := NewNatsRPCServer(cfg, sv, nil, nil)
+	assert.NotNil(t, n.getUserKickChannel())
+}
+
 func TestNatsRPCServerSubscribeToUserMessages(t *testing.T) {
 	cfg := getConfig()
 	sv := getServer()
@@ -374,6 +410,39 @@ func TestNatsRPCServerProcessPushes(t *testing.T) {
 	time.Sleep(30 * time.Millisecond)
 }
 
+func TestNatsRPCServerProcessKick(t *testing.T) {
+	s := helpers.GetTestNatsServer(t)
+	defer s.Shutdown()
+	cfg := viper.New()
+	cfg.Set("pitaya.cluster.rpc.server.nats.connect", fmt.Sprintf("nats://%s", s.Addr()))
+	config := getConfig(cfg)
+	sv := getServer()
+	rpcServer, _ := NewNatsRPCServer(config, sv, nil, nil)
+	err := rpcServer.Init()
+
+	assert.NoError(t, err)
+
+	ctrl := gomock.NewController(t)
+	pitayaSvMock := protosmocks.NewMockPitayaServer(ctrl)
+	defer ctrl.Finish()
+
+	rpcServer.SetPitayaServer(pitayaSvMock)
+
+	kick := &protos.KickMsg{
+		UserId: "someuid",
+	}
+
+	msg, err := proto.Marshal(kick)
+	assert.NoError(t, err)
+
+	pitayaSvMock.EXPECT().KickUser(gomock.Any(), kick).Do(func(ctx context.Context, p *protos.KickMsg) {
+		assert.Equal(t, kick.UserId, p.UserId)
+	})
+
+	rpcServer.userKickCh <- &nats.Msg{Data: msg}
+	time.Sleep(30 * time.Millisecond)
+}
+
 func TestNatsRPCServerReportMetrics(t *testing.T) {
 	cfg := getConfig()
 	sv := getServer()

kick.go

@@ -1,3 +1,22 @@
+// Copyright (c) TFG Co. All Rights Reserved.
+//
+// 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 pitaya
 
 import (
@@ -9,8 +28,8 @@ import (
 	"github.com/topfreegames/pitaya/session"
 )
 
-// SendKickToUser sends kick to an user
-func SendKickToUser(uids []string, frontendType string) error {
+// SendKickToUsers sends kick to an user
+func SendKickToUsers(uids []string, frontendType string) error {
 
 	if !app.server.Frontend && frontendType == "" {
 		return constants.ErrFrontendTypeNotSpecified

kick_test.go

@@ -0,0 +1,96 @@
+// Copyright (c) TFG Co. All Rights Reserved.
+//
+// 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 pitaya
+
+import (
+	"context"
+	"testing"
+
+	"github.com/golang/mock/gomock"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	clustermocks "github.com/topfreegames/pitaya/cluster/mocks"
+	"github.com/topfreegames/pitaya/protos"
+	"github.com/topfreegames/pitaya/session"
+	"github.com/topfreegames/pitaya/session/mocks"
+)
+
+func TestSendKickToUsersLocalSession(t *testing.T) {
+	tables := []struct {
+		name string
+		err  error
+	}{
+		{"success", nil},
+	}
+
+	for _, table := range tables {
+		t.Run(table.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			defer ctrl.Finish()
+			mockNetworkEntity := mocks.NewMockNetworkEntity(ctrl)
+
+			uid1 := uuid.New().String()
+			uid2 := uuid.New().String()
+			s1 := session.New(mockNetworkEntity, true)
+			err := s1.Bind(nil, uid1)
+			assert.NoError(t, err)
+
+			s2 := session.New(mockNetworkEntity, true)
+			err = s2.Bind(nil, uid2)
+			assert.NoError(t, err)
+
+			mockNetworkEntity.EXPECT().Kick(context.Background()).Times(2)
+			mockNetworkEntity.EXPECT().Close().Times(2)
+			err = SendKickToUsers([]string{uid1, uid2}, app.server.Type)
+			assert.NoError(t, err)
+		})
+	}
+}
+
+func TestSendKickToUsersRemoteSession(t *testing.T) {
+	tables := []struct {
+		name string
+	}{
+		{"success"},
+	}
+
+	for _, table := range tables {
+		t.Run(table.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			defer ctrl.Finish()
+			mockRPCClient := clustermocks.NewMockRPCClient(ctrl)
+			app.rpcClient = mockRPCClient
+
+			svType := "connector"
+			uid1 := uuid.New().String()
+			uid2 := uuid.New().String()
+
+			expectedKick1 := &protos.KickMsg{UserId: uid1}
+			expectedKick2 := &protos.KickMsg{UserId: uid2}
+
+			mockRPCClient.EXPECT().SendKick(uid1, gomock.Any(), expectedKick1)
+			mockRPCClient.EXPECT().SendKick(uid2, gomock.Any(), expectedKick2)
+
+			err := SendKickToUsers([]string{uid1, uid2}, svType)
+			assert.NoError(t, err)
+		})
+	}
+}

service/remote_test.go

@@ -189,6 +189,46 @@ func TestRemoteServicePushToUser(t *testing.T) {
 	}
 }
 
+func TestRemoteServiceKickUser(t *testing.T) {
+	svc := NewRemoteService(nil, nil, nil, nil, nil, nil, nil, nil)
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mockNetEntity := sessionmocks.NewMockNetworkEntity(ctrl)
+	tables := []struct {
+		name string
+		uid  string
+		sess *session.Session
+		p    *protos.KickMsg
+		err  error
+	}{
+		{"success", "uid1", session.New(mockNetEntity, true), &protos.KickMsg{
+			UserId: "uid1",
+		}, nil},
+		{"sessionNotFound", "uid2", nil, &protos.KickMsg{
+			UserId: "uid2",
+		}, constants.ErrSessionNotFound},
+	}
+
+	for _, table := range tables {
+		t.Run(table.name, func(t *testing.T) {
+			if table.sess != nil {
+				err := table.sess.Bind(context.Background(), table.uid)
+				assert.NoError(t, err)
+				mockNetEntity.EXPECT().Kick(context.Background())
+				mockNetEntity.EXPECT().Close()
+			}
+			_, err := svc.KickUser(context.Background(), table.p)
+			if table.err != nil {
+				assert.EqualError(t, err, table.err.Error())
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+
+}
+
 func TestRemoteServiceRegisterFailsIfRegisterTwice(t *testing.T) {
 	svc := NewRemoteService(nil, nil, nil, nil, nil, nil, nil, nil)
 	err := svc.Register(&MyComp{}, []component.Option{})