package foundry import ( "context" "errors" "sync" "time" "gitea.local.lab/Lbenedar/foundry_helper_service/internal/foundry/actions" "gitea.local.lab/Lbenedar/foundry_helper_service/internal/foundry/requests" "gitea.local.lab/Lbenedar/foundry_helper_service/internal/foundry/transport" "gitea.local.lab/Lbenedar/foundry_helper_service/internal/foundry/types" ) var ( ListenIsDone = errors.New("Listen for websocket data in foundry is stopped") ReconnectToWebSocket = errors.New("Reconnect to websocket") ChannelIsClosed = errors.New("Channel is closed") CloseTimeoutExceed = errors.New("Timeout of websocket close is exceed") ) type FoundryApi struct { //TODO: make check of admin's authentication transport *transport.FoundryTransport IsAvailable bool // Logger *slog.Logger Status types.FoundryStatus wg sync.WaitGroup } func NewFoundry() *FoundryApi { return &FoundryApi{} } func (foundry *FoundryApi) SetTransport(tr *transport.FoundryTransport) { foundry.transport = tr } func (foundry *FoundryApi) ListenAndServeWS() error { var err error foundry.transport.ReadChan = *types.InitWsChannels() defer foundry.transport.ReadChan.Close() foundry.background(foundry.transport.ListenWebSocket) err = foundry.ServeWebSocket() if err != nil { foundry.IsAvailable = false } return err } func (foundry *FoundryApi) ServeWebSocket() error { var err error wsChannels := &foundry.transport.ReadChan for { select { case message, ok := <-wsChannels.Msg(): if !ok { foundry.transport.Logger.Debug("Read channel is closed", "type", types.WebSocketCode) foundry.transport.Http.SessionID = nil return ChannelIsClosed } foundry.transport.Logger.Debug("Data has been received\n", "msg", message.Code, "type", types.WebSocketCode) //, "msg", message.MsgJson) switch message.Code { case types.RespPingCode, types.RespSessionDataCode: err := foundry.transport.SendOnlyCodeRequest(message.Code) if err != nil { wsChannels.Err() <- &types.FoundryError{Direction: types.WriterCode, Err: err, Type: types.WebSocketCode} continue } case types.RespServerChangeCode: foundry.IsAvailable = true data, err := actions.SplitToTypeAndData([]byte(message.MsgJson)) if err != nil { wsChannels.Err() <- &types.FoundryError{Direction: types.ReaderCode, Err: err, Type: types.WebSocketCode} continue } foundry.transport.Logger.Debug("RespServerChangeCode", "data", data) if data == nil { continue } err = data.Action(foundry.transport, &foundry.Status) if err != nil { wsChannels.Err() <- &types.FoundryError{Direction: types.ReaderCode, Err: err, Type: types.WebSocketCode} continue } case types.RespDataCode: foundry.transport.ExchangeChan.Msgs[message.Id] = make(chan []byte, 1) foundry.transport.ExchangeChan.Msgs[message.Id] <- []byte(message.MsgJson) go foundry.transport.CloseMsgChannel(message.Id, 5*time.Second) default: } case err = <-wsChannels.Err(): var foundryErr *types.FoundryError if errors.As(err, &foundryErr) { if foundryErr.IsFatal { foundry.transport.CloseWebSocketConn() foundry.transport.Http.SessionID = nil return foundryErr } foundry.transport.Logger.Warn("Got error when listening or served", "err", err.Error(), "type", foundryErr.Type, "direction", foundryErr.Direction) } case <-wsChannels.Done(): foundry.transport.CloseWebSocketConn() foundry.transport.Http.SessionID = nil return ListenIsDone case <-wsChannels.Reconnect(): foundry.transport.CloseWebSocketConn() return ReconnectToWebSocket } } } func (foundry *FoundryApi) HandleWSRequest(msgType string) ([]byte, error) { msg := types.NewWsMessageByPage(msgType, foundry.transport.CurrWsId) return foundry.transport.HandleWebsocketRequest(msg) } func (foundry *FoundryApi) Shutdown() error { ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) defer cancel() foundry.transport.Logger.Info("Completing foundry background tasks") foundryClosed := make(chan struct{}) go func() { types.CloseChannel(foundry.transport.ReadChan.Done()) foundry.wg.Wait() foundry.transport.ExchangeChan.Close() foundry.transport.ReadChan.Close() types.CloseChannel(foundryClosed) }() select { case <-foundryClosed: foundry.transport.Logger.Info("Stopped foundry server") case <-ctx.Done(): return CloseTimeoutExceed } return nil } func (foundry *FoundryApi) ConnectToWebSocket() (bool, error) { var err error if !foundry.transport.HasSessionId() { err = foundry.transport.InitSessionId() if err != nil { return false, err } err = foundry.transport.ConnectToFoundry() if err != nil { return false, err } } err = foundry.transport.InitWebSocketConnection() if err != nil { return false, err } status, err := foundry.transport.Http.GetStatus() if err != nil { return false, err } foundry.Status = *types.NewFoundryStatus(status) err = foundry.ListenAndServeWS() if err != nil && err != ListenIsDone { return true, err } return true, nil } func (foundry *FoundryApi) StartListenFoundry() { foundry.background(func() { timeInterval := 1 * time.Second for { ok, err := foundry.ConnectToWebSocket() if err != nil { if errors.Is(err, ReconnectToWebSocket) { if foundry.transport.ReconnectNum >= foundry.transport.ReconnectNumMax { return } time.Sleep(foundry.transport.ReconnectTimeout) foundry.transport.ReconnectNum++ foundry.transport.Logger.Info("Reconnecting to WebSocket", "times", foundry.transport.ReconnectNum) continue } foundry.transport.Logger.Error("Error raised", "err", err.Error()) if ok { timeInterval = 1 * time.Second } foundry.transport.Logger.Info("Trying to reconnect", "timer", timeInterval.String()) time.Sleep(timeInterval) timeInterval = min(timeInterval*2, 15*time.Second) continue } return } }) } func (foundry *FoundryApi) Test() ([]byte, error) { wsMsg := types.NewWsMessage("world", foundry.transport.CurrWsId) foundry.transport.CurrWsId++ answer, err := foundry.transport.HandleWebsocketRequest(wsMsg) if err != nil { return nil, err } // foundry.transport.Logger.Info("World data", "answer", string(answer)) return answer, nil } // TODO: remove func (foundry *FoundryApi) GetHTTP() *requests.FoundryHttpRequest { return foundry.transport.Http }