|
| 1 | +package queue |
| 2 | + |
| 3 | +import ( |
| 4 | + "bytes" |
| 5 | + "encoding/binary" |
| 6 | + "errors" |
| 7 | + "github.com/chenquan/go-pkg/xbinary" |
| 8 | + "github.com/yunqi/lighthouse/internal/packet" |
| 9 | + "github.com/yunqi/lighthouse/internal/persistence/message" |
| 10 | + "github.com/yunqi/lighthouse/internal/persistence/message/encoding" |
| 11 | + "time" |
| 12 | +) |
| 13 | + |
| 14 | +type Message interface { |
| 15 | + Id() packet.Id |
| 16 | + SetId(id packet.Id) |
| 17 | +} |
| 18 | + |
| 19 | +type Publish struct { |
| 20 | + *message.Message |
| 21 | +} |
| 22 | + |
| 23 | +func (p *Publish) Id() packet.Id { |
| 24 | + return p.PacketId |
| 25 | +} |
| 26 | +func (p *Publish) SetId(id packet.Id) { |
| 27 | + p.PacketId = id |
| 28 | +} |
| 29 | + |
| 30 | +type Pubrel struct { |
| 31 | + PacketID packet.Id |
| 32 | +} |
| 33 | + |
| 34 | +func (p *Pubrel) Id() packet.Id { |
| 35 | + return p.PacketID |
| 36 | +} |
| 37 | +func (p *Pubrel) SetId(id packet.Id) { |
| 38 | + p.PacketID = id |
| 39 | +} |
| 40 | + |
| 41 | +// Element represents the element store in the queue. |
| 42 | +type Element struct { |
| 43 | + // At represents the entry time. |
| 44 | + At time.Time |
| 45 | + // Expiry represents the expiry time. |
| 46 | + // Empty means never expire. |
| 47 | + Expiry time.Time |
| 48 | + Message |
| 49 | +} |
| 50 | + |
| 51 | +// Encode encodes the publish structure into bytes and write it to the buffer |
| 52 | +func (p *Publish) Encode(b *bytes.Buffer) { |
| 53 | + encoding.EncodeMessage(p.Message, b) |
| 54 | +} |
| 55 | + |
| 56 | +func (p *Publish) Decode(b *bytes.Buffer) (err error) { |
| 57 | + msg, err := encoding.DecodeMessage(bytes.NewReader(b.Bytes())) |
| 58 | + if err != nil { |
| 59 | + return err |
| 60 | + } |
| 61 | + p.Message = msg |
| 62 | + return nil |
| 63 | +} |
| 64 | + |
| 65 | +// Encode encode the pubrel structure into bytes. |
| 66 | +func (p *Pubrel) Encode(b *bytes.Buffer) { |
| 67 | + _ = xbinary.WriteUint16(b, p.PacketID) |
| 68 | +} |
| 69 | + |
| 70 | +func (p *Pubrel) Decode(b *bytes.Buffer) (err error) { |
| 71 | + p.PacketID, err = xbinary.ReadUint16(b) |
| 72 | + return |
| 73 | +} |
| 74 | + |
| 75 | +// Encode encode the elem structure into bytes. |
| 76 | +// Format: 8 byte timestamp | 1 byte identifier| data |
| 77 | +func (e *Element) Encode() []byte { |
| 78 | + b := bytes.NewBuffer(make([]byte, 0, 100)) |
| 79 | + rs := make([]byte, 19) |
| 80 | + binary.BigEndian.PutUint64(rs[0:9], uint64(e.At.Unix())) |
| 81 | + binary.BigEndian.PutUint64(rs[9:18], uint64(e.Expiry.Unix())) |
| 82 | + switch m := e.Message.(type) { |
| 83 | + case *Publish: |
| 84 | + rs[18] = 0 |
| 85 | + b.Write(rs) |
| 86 | + m.Encode(b) |
| 87 | + case *Pubrel: |
| 88 | + rs[18] = 1 |
| 89 | + b.Write(rs) |
| 90 | + m.Encode(b) |
| 91 | + } |
| 92 | + return b.Bytes() |
| 93 | +} |
| 94 | + |
| 95 | +func (e *Element) Decode(b []byte) (err error) { |
| 96 | + if len(b) < 19 { |
| 97 | + return errors.New("invalid input length") |
| 98 | + } |
| 99 | + e.At = time.Unix(int64(binary.BigEndian.Uint64(b[0:9])), 0) |
| 100 | + e.Expiry = time.Unix(int64(binary.BigEndian.Uint64(b[9:19])), 0) |
| 101 | + switch b[18] { |
| 102 | + case 0: // publish |
| 103 | + p := &Publish{} |
| 104 | + buf := bytes.NewBuffer(b[19:]) |
| 105 | + err = p.Decode(buf) |
| 106 | + e.Message = p |
| 107 | + case 1: // pubrel |
| 108 | + p := &Pubrel{} |
| 109 | + buf := bytes.NewBuffer(b[19:]) |
| 110 | + err = p.Decode(buf) |
| 111 | + e.Message = p |
| 112 | + default: |
| 113 | + return errors.New("invalid identifier") |
| 114 | + } |
| 115 | + return |
| 116 | +} |
0 commit comments