Implement CI mode

Author: g3force

generateProto.sh

@@ -38,6 +38,9 @@ protoc -I"./proto" -I"$GOPATH/src" --go_out="$GOPATH/src" proto/ssl_gc_change.pr
 protoc -I"./proto" -I"$GOPATH/src" --go_out="$GOPATH/src" proto/ssl_gc_api.proto
 protoc -I"./proto" -I"$GOPATH/src" --go_out="$GOPATH/src" proto/ssl_gc_engine.proto
 
+# continuous integration communication
+protoc -I"./proto" -I"$GOPATH/src" --go_out="$GOPATH/src" proto/ssl_gc_ci.proto
+
 # generate javascript code
 pbjs -t static-module -w es6 -o src/proto.js \
   proto/ssl_gc_common.proto \

internal/app/api/server.go

@@ -69,7 +69,7 @@ func (a *Server) WsHandler(w http.ResponseWriter, r *http.Request) {
 }
 
 func (s *ServerConnection) publish() {
-	hook := make(chan *statemachine.StateChange)
+	hook := make(chan engine.HookOut)
 	s.gcEngine.RegisterHook(hook)
 	defer func() {
 		s.gcEngine.UnregisterHook(hook)
@@ -84,14 +84,16 @@ func (s *ServerConnection) publish() {
 		select {
 		case <-s.quit:
 			return
-		case change := <-hook:
-			s.publishState(change.State)
-			s.publishProtocolDelta()
-		case <-time.After(100 * time.Millisecond):
-			s.publishGcState()
-			s.publishState(s.gcEngine.CurrentState())
-			if s.gcEngine.LatestChangeId() != s.lastProtocolId {
-				s.publishProtocolFull()
+		case hookOut := <-hook:
+			if hookOut.Change != nil {
+				s.publishState(hookOut.State)
+				s.publishProtocolDelta()
+			} else if hookOut.State != nil {
+				s.publishGcState()
+				s.publishState(hookOut.State)
+				if s.gcEngine.LatestChangeId() != s.lastProtocolId {
+					s.publishProtocolFull()
+				}
 			}
 		}
 	}

internal/app/ci/ci.go

@@ -0,0 +1,107 @@
+package ci
+
+import (
+	"github.com/RoboCup-SSL/ssl-game-controller/internal/app/state"
+	"github.com/RoboCup-SSL/ssl-game-controller/internal/app/tracker"
+	"github.com/RoboCup-SSL/ssl-go-tools/pkg/sslconn"
+	"log"
+	"net"
+	"time"
+)
+
+type Handler func(time.Time, *tracker.TrackerWrapperPacket) *state.Referee
+
+type Server struct {
+	address         string
+	listener        net.Listener
+	conn            net.Conn
+	latestTime      time.Time
+	tickChan        chan time.Time
+	TrackerConsumer func(*tracker.TrackerWrapperPacket)
+}
+
+func NewServer(address string) *Server {
+	return &Server{address: address, tickChan: make(chan time.Time, 1)}
+}
+
+func (s *Server) Start() {
+	go s.listen(s.address)
+}
+
+func (s *Server) Stop() {
+	if err := s.listener.Close(); err != nil {
+		log.Printf("Could not close listener: %v", err)
+	}
+}
+
+func (s *Server) listen(address string) {
+	listener, err := net.Listen("tcp", address)
+	if err != nil {
+		log.Print("Failed to listen on ", address)
+		return
+	}
+	log.Print("Listening on ", address)
+	s.listener = listener
+
+	for {
+		conn, err := listener.Accept()
+		if err != nil {
+			log.Print("Could not accept connection: ", err)
+		} else {
+			log.Println("CI connection established")
+			s.conn = conn
+			s.serve(conn)
+			log.Println("CI connection closed")
+		}
+	}
+}
+
+func (s *Server) serve(conn net.Conn) {
+	defer func() {
+		if err := conn.Close(); err != nil {
+			log.Printf("Could not close CI client connection: %v", err)
+		}
+	}()
+
+	for {
+		input := CiInput{}
+		if err := sslconn.ReceiveMessage(conn, &input); err != nil {
+			log.Println("Could not receive message from CI connection: ", err)
+			return
+		}
+
+		if input.TrackerPacket != nil {
+			s.TrackerConsumer(input.TrackerPacket)
+		}
+
+		if input.Timestamp != nil {
+			sec := *input.Timestamp / 1e9
+			nSec := *input.Timestamp - sec*1e9
+			s.latestTime = time.Unix(sec, nSec)
+			select {
+			case s.tickChan <- time.Now():
+			default:
+			}
+		}
+	}
+}
+
+func (s *Server) SendMessage(refMsg *state.Referee) {
+	if s.conn == nil {
+		log.Println("Could not send message to CI client: Not connected")
+		return
+	}
+	output := CiOutput{RefereeMsg: refMsg}
+	if err := sslconn.SendMessage(s.conn, &output); err != nil {
+		log.Printf("Could not send message: %v", err)
+		return
+	}
+}
+
+func (s *Server) Time() time.Time {
+	return s.latestTime
+}
+
+func (s *Server) TickChanProvider() <-chan time.Time {
+	return s.tickChan
+}

internal/app/ci/ssl_gc_ci.pb.go

@@ -23,13 +23,14 @@ type Engine struct {
 	currentState             *state.State
 	stateMachine             *statemachine.StateMachine
 	queue                    chan *statemachine.Change
-	hooks                    []chan *statemachine.StateChange
+	hooks                    []chan HookOut
 	timeProvider             timer.TimeProvider
 	lastTimeUpdate           time.Time
 	gcState                  *GcState
 	gcStateMutex             sync.Mutex
 	noProgressDetector       NoProgressDetector
 	ballPlacementCoordinator BallPlacementCoordinator
+	tickChanProvider         func() <-chan time.Time
 }
 
 // NewEngine creates a new engine
@@ -39,9 +40,8 @@ func NewEngine(gameConfig config.Game) (e *Engine) {
 	e.stateStore = store.NewStore(gameConfig.StateStoreFile)
 	e.stateMachine = statemachine.NewStateMachine(gameConfig)
 	e.queue = make(chan *statemachine.Change, 100)
-	e.hooks = []chan *statemachine.StateChange{}
-	e.timeProvider = func() time.Time { return time.Now() }
-	e.lastTimeUpdate = e.timeProvider()
+	e.hooks = []chan HookOut{}
+	e.SetTimeProvider(func() time.Time { return time.Now() })
 	e.gcState = new(GcState)
 	e.gcState.TeamState = map[string]*GcStateTeam{
 		state.Team_YELLOW.String(): new(GcStateTeam),
@@ -51,6 +51,9 @@ func NewEngine(gameConfig config.Game) (e *Engine) {
 	e.gcState.TrackerStateGc = new(GcStateTracker)
 	e.noProgressDetector = NoProgressDetector{gcEngine: e}
 	e.ballPlacementCoordinator = BallPlacementCoordinator{gcEngine: e}
+	e.tickChanProvider = func() <-chan time.Time {
+		return time.After(25 * time.Millisecond)
+	}
 	return
 }
 
@@ -64,26 +67,6 @@ func (e *Engine) Enqueue(change *statemachine.Change) {
 	e.queue <- change
 }
 
-// RegisterHook registers given hook for post processing after each change
-func (e *Engine) RegisterHook(hook chan *statemachine.StateChange) {
-	e.hooks = append(e.hooks, hook)
-}
-
-// UnregisterHook unregisters hooks that were registered before
-func (e *Engine) UnregisterHook(hook chan *statemachine.StateChange) bool {
-	for i, h := range e.hooks {
-		if h == hook {
-			e.hooks = append(e.hooks[:i], e.hooks[i+1:]...)
-			select {
-			case <-hook:
-			case <-time.After(10 * time.Millisecond):
-			}
-			return true
-		}
-	}
-	return false
-}
-
 // SetGeometry sets a new geometry
 func (e *Engine) SetGeometry(geometry config.Geometry) {
 	e.stateMachine.Geometry = geometry
@@ -94,6 +77,17 @@ func (e *Engine) GetGeometry() config.Geometry {
 	return e.stateMachine.Geometry
 }
 
+// SetTimeProvider sets a new time provider for this engine
+func (e *Engine) SetTimeProvider(provider timer.TimeProvider) {
+	e.timeProvider = provider
+	e.lastTimeUpdate = e.timeProvider()
+}
+
+// SetTickChanProvider sets an alternative provider for the tick channel
+func (e *Engine) SetTickChanProvider(provider func() <-chan time.Time) {
+	e.tickChanProvider = provider
+}
+
 // Start loads the state store and runs a go routine that consumes the change queue
 func (e *Engine) Start() error {
 	if err := e.stateStore.Open(); err != nil {
@@ -166,13 +160,8 @@ func (e *Engine) processChanges() {
 				return
 			}
 			e.processChange(change)
-		case <-time.After(10 * time.Millisecond):
+		case <-e.tickChanProvider():
 			e.processTick()
-
-			e.noProgressDetector.process()
-			e.ballPlacementCoordinator.process()
-			e.processPrepare()
-			e.processBotNumber()
 		}
 	}
 }
@@ -240,7 +229,7 @@ func (e *Engine) processChange(change *statemachine.Change) {
 		log.Println("Could not add new state to store: ", err)
 	}
 	for _, hook := range e.hooks {
-		hook <- &entry
+		hook <- HookOut{Change: entry.Change, State: e.CurrentState()}
 	}
 }