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                         Metadata:   command.Metadata,
  26                         Capability: command.Capability,
  27                         Payload:    command.Payload,
  28                 })
  29         }
  30
  31         data, err := json.Marshal(batchWirePayload{Commands: commands})
  32         if err != nil {
  33                 return nil, fmt.Errorf("encode batch members: %w", err)
  34         }
  35
  36         return data, nil
  37 }
  38
  39 // UnmarshalJSON rejects unknown fields and unsupported member schemas.
  40 func (b *BatchPayload) UnmarshalJSON(data []byte) error {
  41         var wire batchWirePayload
  42
  43         err := decodeObject(data, &wire)
  44         if err != nil {
  45                 return fmt.Errorf("decode batch members: %w", err)
  46         }
  47
  48         commands := make([]BatchCommand, 0, len(wire.Commands))
  49
  50         for _, command := range wire.Commands {
  51                 if command.Schema != Schema {
  52                         return fmt.Errorf("unsupported batch member schema %d", command.Schema)
  53                 }
  54
  55                 commands = append(commands, BatchCommand{
  56                         Metadata:   command.Metadata,
  57                         Capability: command.Capability,
  58                         Payload:    command.Payload,
  59                 })
  60         }
  61
  62         b.Commands = commands
  63
  64         return nil
  65 }
  66
  67 // BatchCommand holds one complete request without a topic or schema field. The
  68 // outer batch message ID equals the first member's message ID.
  69 type BatchCommand struct {
  70         Metadata
  71         Capability string
  72         Payload    ValuePayload
  73 }
  74
  75 type batchWirePayload struct {
  76         Commands []batchWireCommand `json:"commands"`
  77 }
  78
  79 type batchWireCommand struct {
  80         Schema int `json:"schema"`
  81         Metadata
  82         Capability string       `json:"capability"`
  83         Payload    ValuePayload `json:"payload"`
  84 }