func (c *Conn) handler(pkt lp.Packet) error {
	start := time.Now()
	var status int = 200
	defer func() {
		c.service.meter.ConnTimeSeries.AddTime(time.Since(start))
		c.service.stats.PrecisionTiming("conn_time_ns", time.Since(start), stats.IntTag("status", status))
	}()
	c.storeInbound(pkt)
	switch pkt.Type() {
	case lp.CONNECT:
		var returnCode uint8
		packet := *pkt.(*lp.Connect)
		c.insecure = packet.InsecureFlag
		c.username = string(packet.Username)
		clientid, err := c.onConnect(packet.ClientID)
		if err != nil {
			status = err.Status
			c.notifyError(err, 0)
			c.sendClientID(clientid.Encode(c.service.MAC))
			returnCode = 0x05  
		}
		c.clientid = clientid
		c.MessageIds.Reset(message.MID(c.connid))
		if !packet.CleanSessFlag {
			c.resume()
		} else {
			store.Log.Reset(c.clientid.Contract())
		}
		connack := &lp.Connack{ReturnCode: returnCode, ConnID: uint32(c.connid)}
		c.send <- connack
	case lp.SUBSCRIBE:
		packet := *pkt.(*lp.Subscribe)
		ack := &lp.Suback{
			MessageID: packet.MessageID,
			Qos:       make([]uint8, 0, len(packet.Subscriptions)),
		}
		for _, sub := range packet.Subscriptions {
			if err := c.onSubscribe(packet, sub.Topic); err != nil {
				status = err.Status
				ack.Qos = append(ack.Qos, 0x80)  
				c.notifyError(err, packet.MessageID)
				continue
			}
			ack.Qos = append(ack.Qos, sub.Qos)
		}
		if packet.IsForwarded {
			return nil
		}
		c.send <- ack
	case lp.UNSUBSCRIBE:
		packet := *pkt.(*lp.Unsubscribe)
		ack := &lp.Unsuback{MessageID: packet.MessageID}
		for _, sub := range packet.Subscriptions {
			if err := c.onUnsubscribe(packet, sub.Topic); err != nil {
				status = err.Status
				c.notifyError(err, packet.MessageID)
			}
		}
		c.send <- ack
	case lp.PINGREQ:
		resp := &lp.Pingresp{}
		c.send <- resp
	case lp.DISCONNECT:
	case lp.PUBLISH:
		packet := *pkt.(*lp.Publish)
		if err := c.onPublish(packet, packet.MessageID, packet.Topic, packet.Payload); err != nil {
			status = err.Status
			c.notifyError(err, packet.MessageID)
		}
	case lp.PUBREC:
		packet := *pkt.(*lp.Pubrec)
		pubrel := &lp.Pubrel{
			FixedHeader: lp.FixedHeader{
				Qos: packet.Qos,
			},
			MessageID: packet.MessageID,
		}
		c.send <- pubrel
	case lp.PUBREL:
		c.storeOutbound(pkt)
		packet := *pkt.(*lp.Pubrel)
		pubcomp := &lp.Pubcomp{MessageID: packet.MessageID}
		c.send <- pubcomp
	case lp.PUBCOMP:
	}
	return nil
}
handle handles inbound packets.
package broker
import (
	"bufio"
	"context"
	"encoding/json"
	"time"
	lp "github.com/unit-io/unitd/lineprotocol"
	"github.com/unit-io/unitd/message"
	"github.com/unit-io/unitd/message/security"
	"github.com/unit-io/unitd/pkg/crypto"
	"github.com/unit-io/unitd/pkg/log"
	"github.com/unit-io/unitd/pkg/stats"
	"github.com/unit-io/unitd/pkg/uid"
	"github.com/unit-io/unitd/store"
	"github.com/unit-io/unitd/types"
)
const (
	requestClientId = 2682859131  
	requestKeygen   = 812942072   
)
func (c *Conn) readLoop() error {
	defer func() {
		log.Info("conn.Handler", "closing...")
		c.close()
	}()
	reader := bufio.NewReaderSize(c.socket, 65536)
	for {
		c.socket.SetDeadline(time.Now().Add(time.Second * 120))
		pkt, err := lp.ReadPacket(c.proto, reader)
		if err != nil {
			return err
		}
		if err := c.handler(pkt); err != nil {
			return err
		}
	}
}