subscribe.go 2.1 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485
  1. /*
  2. * Copyright (c) 2021 IBM Corp and others.
  3. *
  4. * All rights reserved. This program and the accompanying materials
  5. * are made available under the terms of the Eclipse Public License v2.0
  6. * and Eclipse Distribution License v1.0 which accompany this distribution.
  7. *
  8. * The Eclipse Public License is available at
  9. * https://www.eclipse.org/legal/epl-2.0/
  10. * and the Eclipse Distribution License is available at
  11. * http://www.eclipse.org/org/documents/edl-v10.php.
  12. *
  13. * Contributors:
  14. * Allan Stockdill-Mander
  15. */
  16. package packets
  17. import (
  18. "bytes"
  19. "fmt"
  20. "io"
  21. )
  22. // SubscribePacket is an internal representation of the fields of the
  23. // Subscribe MQTT packet
  24. type SubscribePacket struct {
  25. FixedHeader
  26. MessageID uint16
  27. Topics []string
  28. Qoss []byte
  29. }
  30. func (s *SubscribePacket) String() string {
  31. return fmt.Sprintf("%s MessageID: %d topics: %s", s.FixedHeader, s.MessageID, s.Topics)
  32. }
  33. func (s *SubscribePacket) Write(w io.Writer) error {
  34. var body bytes.Buffer
  35. var err error
  36. body.Write(encodeUint16(s.MessageID))
  37. for i, topic := range s.Topics {
  38. body.Write(encodeString(topic))
  39. body.WriteByte(s.Qoss[i])
  40. }
  41. s.FixedHeader.RemainingLength = body.Len()
  42. packet := s.FixedHeader.pack()
  43. packet.Write(body.Bytes())
  44. _, err = packet.WriteTo(w)
  45. return err
  46. }
  47. // Unpack decodes the details of a ControlPacket after the fixed
  48. // header has been read
  49. func (s *SubscribePacket) Unpack(b io.Reader) error {
  50. var err error
  51. s.MessageID, err = decodeUint16(b)
  52. if err != nil {
  53. return err
  54. }
  55. payloadLength := s.FixedHeader.RemainingLength - 2
  56. for payloadLength > 0 {
  57. topic, err := decodeString(b)
  58. if err != nil {
  59. return err
  60. }
  61. s.Topics = append(s.Topics, topic)
  62. qos, err := decodeByte(b)
  63. if err != nil {
  64. return err
  65. }
  66. s.Qoss = append(s.Qoss, qos)
  67. payloadLength -= 2 + len(topic) + 1 // 2 bytes of string length, plus string, plus 1 byte for Qos
  68. }
  69. return nil
  70. }
  71. // Details returns a Details struct containing the Qos and
  72. // MessageID of this ControlPacket
  73. func (s *SubscribePacket) Details() Details {
  74. return Details{Qos: 1, MessageID: s.MessageID}
  75. }