func (c *Conn) onPublish(pkt lp.Publish, messageID uint16, msgTopic []byte, payload []byte) *types.Error {
	start := time.Now()
	defer log.ErrLogger.Debug().Str("context", "conn.onPublish").Int64("duration", time.Since(start).Nanoseconds()).Msg("")
	topic := security.ParseKey(msgTopic)
	if topic.TopicType == security.TopicInvalid {
		return types.ErrBadRequest
	}
	if len(topic.Key) == 5 && string(topic.Key) == "unitd" {
		c.onUnitdRequest(topic, payload)
		return nil
	}
	if !c.insecure {
		wildcard, err := c.onSecureRequest(topic)
		if err != nil {
			return err
		}
		if wildcard {
			return types.ErrForbidden
		}
	}
	err := store.Message.Put(c.clientid.Contract(), topic.Topic, payload)
	if err != nil {
		log.Error("conn.onPublish", "store message "+err.Error())
		return types.ErrServerError
	}
	c.storeOutbound(&pkt)
	c.publish(pkt, messageID, topic, payload)
	return c.ack(pkt)
}
OnPublish is a handler for Publish events.
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
		}
	}
}
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
}
func (c *Conn) writeLoop(ctx context.Context) {
	c.closeW.Add(1)
	defer c.closeW.Done()
	for {
		select {
		case <-ctx.Done():
			return
		case <-c.closeC:
			return
		case msg, ok := <-c.pub:
			if !ok {
				return
			}
			m, err := lp.Encode(c.proto, msg)
			if err != nil {
				log.Error("conn.writeLoop", err.Error())
				return
			}
			c.socket.Write(m.Bytes())
		case msg, ok := <-c.send:
			if !ok {
				return
			}
			m, err := lp.Encode(c.proto, msg)
			if err != nil {
				log.Error("conn.writeLoop", err.Error())
				return
			}
			c.socket.Write(m.Bytes())
		}
	}
}
func (c *Conn) onConnect(clientID []byte) (uid.ID, *types.Error) {
	start := time.Now()
	defer log.ErrLogger.Debug().Str("context", "conn.onConnect").Int64("duration", time.Since(start).Nanoseconds()).Msg("")
	var clientid = uid.ID{}
	if clientID != nil && len(clientID) > c.service.MAC.Overhead() {
		if contract, ok := c.service.cache.Load(crypto.SignatureToUint32(clientID[crypto.EpochSize:crypto.MessageOffset])); ok {
			clientid, err := uid.CachedClientID(contract.(uint32))
			if err != nil {
				return nil, types.ErrUnauthorized
			}
			return clientid, nil
		}
	}
	clientid, err := uid.Decode(clientID, c.service.MAC)
	if err != nil {
		clientid, err = uid.NewClientID(1)
		if err != nil {
			return nil, types.ErrUnauthorized
		}
		return clientid, types.ErrInvalidClientId
	}
	if !clientid.IsPrimary() {
		cid := []byte(clientid.Encode(c.service.MAC))
		c.service.cache.LoadOrStore(crypto.SignatureToUint32(cid[crypto.EpochSize:crypto.MessageOffset]), clientid.Contract())
	}
	return clientid, nil
}
func (c *Conn) onSubscribe(pkt lp.Subscribe, msgTopic []byte) *types.Error {
	start := time.Now()
	defer log.ErrLogger.Debug().Str("context", "conn.onSubscribe").Int64("duration", time.Since(start).Nanoseconds()).Msg("")
	topic := security.ParseKey(msgTopic)
	if topic.TopicType == security.TopicInvalid {
		return types.ErrBadRequest
	}
	if !c.insecure {
		if _, err := c.onSecureRequest(topic); err != nil {
			return err
		}
	}
	c.storeOutbound(&pkt)
	c.subscribe(pkt, topic)
	msgs, err := store.Message.Get(c.clientid.Contract(), topic.Topic)
	if err != nil {
		log.Error("conn.OnSubscribe", "query last messages"+err.Error())
		return types.ErrServerError
	}
	for _, m := range msgs {
		msg := m  
		c.SendMessage(&msg)
	}
	return nil
}
func (c *Conn) onUnsubscribe(pkt lp.Unsubscribe, msgTopic []byte) *types.Error {
	start := time.Now()
	defer log.ErrLogger.Debug().Str("context", "conn.onUnsubscribe").Int64("duration", time.Since(start).Nanoseconds()).Msg("")
	topic := security.ParseKey(msgTopic)
	if topic.TopicType == security.TopicInvalid {
		return types.ErrBadRequest
	}
	if !c.insecure {
		if _, err := c.onSecureRequest(topic); err != nil {
			return err
		}
	}
	c.storeOutbound(&pkt)
	c.unsubscribe(pkt, topic)
	return nil
}