package session import ( "fmt" "log" "badat.dev/maeqtt/v2/mqtt/packets" ) type Session struct { ClientID *string // Nullable Connection *Connection SubscriptionChannel chan packets.PublishPacket ConnecionChannel chan ConnectionRequest freePacketID uint16 Expiry } func NewSession(req ConnectionRequest, rmSessChan RemoveSessionChannel) Session { sess := Session{} sess.SubscriptionChannel = make(chan packets.PublishPacket,) sess.Expiry = NewExpiry(rmSessChan) sess.Connect(req) return sess } func (s *Session) Connect(req ConnectionRequest) { if s.Connection != nil { s.Disconnect(packets.DisconnectReasonCodeSessionTakenOver) } connAck := packets.ConnackPacket{} s.SetExpireTimer(req.ConnectPakcet.Properties.SessionExpiryInterval.Value) s.expireTimer.Stop() if req.ConnectPakcet.ClientId == nil { if s.ClientID == nil { s.ClientID = genClientID() } connAck.Properties.AssignedClientIdentifier.Value = s.ClientID } else if s.ClientID != nil && s.ClientID != req.ConnectPakcet.ClientId { panic(fmt.Errorf("Session %s connect called with a connect packet with an ID: %s", *s.ClientID, *req.ConnectPakcet.ClientId)) } else { s.ClientID = req.ConnectPakcet.ClientId } true := byte(1) false := byte(0) connAck.Properties.WildcardSubscriptionAvailable.Value = &true connAck.Properties.RetainAvailable.Value = &false connAck.Properties.SharedSubscriptionAvailable.Value = &false s.Connection = req.Connection err := s.Connection.sendPacket(connAck) if err != nil { panic("TODO, handle this") } } // Starts a loop the receives and responds to packets func (s *Session) HandlerLoop() { go s.Connection.PacketReadLoop() for s.Connection != nil { select { case packet := <-s.Connection.PacketChannel: packet.Visit(s) case <-s.Connection.ClientDisconnectedChan: s.onDisconnect() case c := <-s.ConnecionChannel: s.Connect(c) case subMessage := <-s.SubscriptionChannel: // TODO implement other qos levels subMessage.QOSLevel = 0 subMessage.Dup = false err := s.Connection.sendPacket(subMessage) if err != nil { panic("TOOO handle this") } } } select { case c := <-s.ConnecionChannel: s.Connect(c) // Tail recursion baybeeee s.HandlerLoop() case <- s.expireTimer.C: s.expireSession() } } func (s *Session) onDisconnect() { s.Connection = nil s.resetExpireTimer() log.Printf("Client disconnected, id: %s", *s.ClientID) }