feat: support skcp

controllers/helper.go

@@ -77,6 +77,7 @@ const DiffNew = "New"
 const DiffUpdated = "Updated"
 const DiffDeleted = "Deleted"
 const DiffUnchanged = "Unchanged"
+const DiffRoomUpdated = "RoomUpdated"
 
 type RoomKey struct {
 	ServerId string
@@ -174,11 +175,7 @@ func DiffRoomStatus(past *v1.RoomIngressStatus, curr *v1.RoomIngressStatus, opts
 			diff.Type = DiffNew
 			diff.Current = c
 		} else if cok && pok {
-			if o.uh(past, p, curr, c) {
-				diff.Type = DiffUpdated
-			} else {
-				diff.Type = DiffUnchanged
-			}
+			diff.Type = o.uh(past, p, curr, c)
 			diff.Current = c
 			diff.Past = p
 		} else if !cok && pok {
@@ -193,23 +190,29 @@ func DiffRoomStatus(past *v1.RoomIngressStatus, curr *v1.RoomIngressStatus, opts
 }
 
 func TokenUpdatedHandler() UpdatedHandler {
-	return func(past *v1.RoomIngressStatus, pp PlayerPos, curr *v1.RoomIngressStatus, cp PlayerPos) bool {
+	return func(past *v1.RoomIngressStatus, pp PlayerPos, curr *v1.RoomIngressStatus, cp PlayerPos) DiffType {
 		p := GetPlayerStatusByPos(past, pp)
 		c := GetPlayerStatusByPos(curr, cp)
 		if c.Player.Status == v1.PlayerStatusRetry {
-			return true
+			return DiffUpdated
+		}
+		if p.Room.Upstream != c.Room.Upstream {
+			return DiffRoomUpdated
+		}
+		if p.Player.Token != c.Player.Token {
+			return DiffUpdated
 		}
-		return p.Player.Token != c.Player.Token
+		return DiffUnchanged
 	}
 }
 
 func AlwaysUpdatedHandler() UpdatedHandler {
-	return func(past *v1.RoomIngressStatus, pp PlayerPos, curr *v1.RoomIngressStatus, cp PlayerPos) bool {
-		return true
+	return func(past *v1.RoomIngressStatus, pp PlayerPos, curr *v1.RoomIngressStatus, cp PlayerPos) DiffType {
+		return DiffUpdated
 	}
 }
 
-type UpdatedHandler func(past *v1.RoomIngressStatus, pp PlayerPos, curr *v1.RoomIngressStatus, cp PlayerPos) bool
+type UpdatedHandler func(past *v1.RoomIngressStatus, pp PlayerPos, curr *v1.RoomIngressStatus, cp PlayerPos) DiffType
 
 type diffOptions struct {
 	uh UpdatedHandler

controllers/roomingress.go

@@ -159,6 +159,9 @@ func (it *RoomIngressReconciler) syncTokens(ring *v1.RoomIngress) (int, *time.Du
 				}
 			}
 		}
+		if diff.Type == DiffRoomUpdated {
+			n++
+		}
 	}
 	ring.Status = *curr
 	return n, min(requeue)

controllers/roomingress_test.go

@@ -93,4 +93,13 @@ func TestRoomIngressController(t *testing.T) {
 	p4 = GetPlayerStatusByKey(&ring1.Status, PlayerKey{RoomKey: room11, PlayerId: "player4"})
 	assert.Equal(t, v1.PlayerStatusExpired, p1.Player.Status)
 	assert.Nil(t, p4)
+
+	ring1.Spec.Rooms[0].Upstream = "127.0.0.2:4321"
+	n, u = rec.syncTokens(ring1)
+	assert.Equal(t, 3, n)
+	// _ = n
+	assert.Equal(t, "127.0.0.2:4321", ring1.Status.Rooms[0].Upstream)
+
+	n, _ = rec.syncTokens(ring1)
+	assert.Equal(t, 0, n)
 }

protocol/multiplex/parser.go

@@ -18,6 +18,8 @@ func Parser() protocol.FuncMultiplexParser {
 			fallthrough
 		case 'e':
 			fallthrough
+		case 's':
+			fallthrough
 		case 'x':
 			return b, nil
 		default:

protocol/skcp/parser.go

@@ -0,0 +1,20 @@
+package skcp
+
+import (
+	"errors"
+	"encoding/binary"
+
+	"github.com/LilithGames/spiracle/protocol"
+)
+
+
+func Parser() protocol.FuncTokenParser {
+	size := 25
+	return func(data []byte) (uint32, error) {
+		if len(data) < size {
+			return 0, errors.New("invalid skcp data")
+		}
+		token := binary.LittleEndian.Uint32(data[1:])
+		return token, nil
+	}
+}

protocol/skcp/parser_test.go

@@ -0,0 +1,17 @@
+package skcp
+
+import (
+	"testing"
+
+	assert "github.com/stretchr/testify/require"
+)
+
+
+func TestSKcp(t *testing.T) {
+	p := Parser()
+	buffer := make([]byte, 26)
+	buffer[2] = 0x01
+	token, err := p(buffer[1:])
+	assert.Nil(t, err)
+	assert.Equal(t, uint32(1), token)
+}

services/roomproxy/handler.go

@@ -31,6 +31,8 @@ func (it *RoomProxy) multiplexing(buffer []byte) proxyHandler {
 		return proxyHandler{ch: ch, u: it.ukcp, d: it.dkcp}
 	case 'e':
 		return proxyHandler{ch: ch, u: it.uecho, d: it.decho}
+	case 's':
+		return proxyHandler{ch: ch, u: it.ukcp, d: it.dkcp}
 	default:
 		return proxyHandler{ch: ch, u: it.drop, d: it.drop}
 	}
@@ -51,6 +53,12 @@ func (it *RoomProxy) kcptoken(ch byte, buffer []byte) (uint32, error) {
 			return 0, fmt.Errorf("heartbeat.GetToken err: %w", err)
 		}
 		return token, nil
+	case 's':
+		token, err := it.skcp.GetToken(buffer[1:])
+		if err != nil {
+			return 0, fmt.Errorf("skcp.GetToken err: %w", err)
+		}
+		return token, nil
 	default:
 		return 0, errors.New("unknown kcp channel")
 	}

services/roomproxy/service.go

@@ -2,13 +2,14 @@ package roomproxy
 
 import (
 	"context"
-	"time"
 	"log"
+	"time"
 
 	"github.com/LilithGames/spiracle/protocol"
 	"github.com/LilithGames/spiracle/protocol/heartbeat"
 	"github.com/LilithGames/spiracle/protocol/kcp"
 	"github.com/LilithGames/spiracle/protocol/multiplex"
+	"github.com/LilithGames/spiracle/protocol/skcp"
 	"github.com/LilithGames/spiracle/proxy"
 	"github.com/LilithGames/spiracle/repos"
 )
@@ -19,6 +20,7 @@ type RoomProxy struct {
 	name      string
 	multiplex protocol.MultiplexParser
 	kcp       protocol.TokenParser
+	skcp      protocol.TokenParser
 	heartbeat protocol.TokenParser
 }
 
@@ -47,6 +49,7 @@ func NewRoomProxy(ctx context.Context, name string, opts ...RoomProxyOption) (*R
 		name:             name,
 		multiplex:        protocol.NewFuncMultiplexParser(multiplex.Parser()),
 		kcp:              protocol.NewFuncTokenParser(kcp.Parser()),
+		skcp:             protocol.NewFuncTokenParser(skcp.Parser()),
 		heartbeat:        protocol.NewFuncTokenParser(heartbeat.Parser()),
 	}
 	return rp, nil

services/roomproxy/service_test.go

@@ -47,7 +47,7 @@ func client(t *testing.T) *net.UDPConn {
 
 func TestRoomProxyReal(t *testing.T) {
 	s := &proxy.Statd{}
-	go s.Tick(nil)
+	go s.Tick(func(s *proxy.Statd){})
 	ctx := proxy.WithStatd(context.TODO(), s)
 	name := "server1"
 	roomproxy, err := NewRoomProxy(ctx, name)