git.giacomolli.uk - onehouse/blob - protocol/batch_payload.go

   1 package protocol
   2
   3 import (
   4         "encoding/json"
   5         "fmt"
   6 )
   7
   8 // MaximumCommandBatchSize bounds the ordered commands in one publication.
   9 const MaximumCommandBatchSize = 256
  10
  11 // BatchPayload holds 2 through 256 commands for one device. IDs and targets are
  12 // distinct. Array order is operation order. Each member keeps its own expiry,
  13 // correlation, and causation. There is no combined result.
  14 type BatchPayload struct {
  15         Commands []BatchCommand
  16 }
  17
  18 // MarshalJSON inserts each command's schema without caller-built envelopes.
  19 func (b BatchPayload) MarshalJSON() ([]byte, error) {
  20         commands := make([]batchWireCommand, 0, len(b.Commands))
  21
  22         for _, command := range b.Commands {
  23                 commands = append(commands, batchWireCommand{
  24                         Schema: Schema,
  25
  26                         Metadata: command.Metadata,
  27
  28                         Capability: command.Capability,
  29
  30                         Payload: command.Payload,
  31                 })
  32         }
  33
  34         data, err := json.Marshal(batchWirePayload{Commands: commands})
  35         if err != nil {
  36                 return nil, fmt.Errorf("encode batch members: %w", err)
  37         }
  38
  39         return data, nil
  40 }
  41
  42 // UnmarshalJSON rejects unknown fields and unsupported member schemas.
  43 func (b *BatchPayload) UnmarshalJSON(data []byte) error {
  44         var wire batchWirePayload
  45
  46         err := decodeObject(data, &wire)
  47         if err != nil {
  48                 return fmt.Errorf("decode batch members: %w", err)
  49         }
  50
  51         commands := make([]BatchCommand, 0, len(wire.Commands))
  52
  53         for _, command := range wire.Commands {
  54                 if command.Schema != Schema {
  55                         return fmt.Errorf("unsupported batch member schema %d", command.Schema)
  56                 }
  57
  58                 commands = append(commands, BatchCommand{
  59                         Metadata: command.Metadata,
  60
  61                         Capability: command.Capability,
  62
  63                         Payload: command.Payload,
  64                 })
  65         }
  66
  67         b.Commands = commands
  68
  69         return nil
  70 }
  71
  72 // BatchCommand holds one complete request without a topic or schema field. The
  73 // outer batch message ID equals the first member's message ID.
  74 type BatchCommand struct {
  75         Metadata
  76         Capability string
  77         Payload    ValuePayload
  78 }
  79
  80 type batchWirePayload struct {
  81         Commands []batchWireCommand `json:"commands"`
  82 }
  83
  84 type batchWireCommand struct {
  85         Schema int `json:"schema"`
  86         Metadata
  87         Capability string       `json:"capability"`
  88         Payload    ValuePayload `json:"payload"`
  89 }