std/amqptest

struct QueueOptions {
  sessions: Bool = false
  lock_ms: Int = 60000
  max_delivery_count: Int = 10
}

struct FakeMsg {
  seq: Int
  m: amqp.Message
  delivery_count: Int = 0
  lock_token: Bytes = b""
  locked_until: Int = 0
  lock_conn: Int = 0
  scheduled_ms: Int = 0
  enqueued_ms: Int = 0
  session: Str = ""
  deferred: Bool = false
  dead_reason: Str = ""
  dead_description: Str = ""
}

struct FakeSession {
  id: Str
  state: Bytes = b""
  locked_until: Int = 0
  owner: Int = 0
  updated_ms: Int = 0
}

struct FakeQueue {
  name: Str
  opts: QueueOptions
  is_topic: Bool = false
  msgs: List[FakeMsg] = []
  dead: List[FakeMsg] = []
  next_seq: Int = 1
  sessions: List[FakeSession] = []
}

struct FakeFail {
  op: Str
  status: Int
  condition: Str
  description: Str
  left: Int
}

struct FakePart {
  events: List[amqp.Message] = []
}

struct FakeHub {
  name: Str
  groups: List[Str]
  parts: List[FakePart]
  created_ms: Int
  rr: Int = 0
  geo_factor: Int = 1
}

struct HubOwner {
  path: Str
  key: Str
  epoch: Int
}

struct BrokerState {
  queues: List[FakeQueue] = []
  hubs: List[FakeHub] = []
  owners: List[HubOwner] = []
  tokens: List[(Str, Str)] = []
  attaches: List[(Str, Str, List[Str])] = []
  expiries: List[(Str, Int)] = []
  rejected: List[Str] = []
  require_token: Bool = false
  fails: List[FakeFail] = []
  wakes: List[(Int, Chan[Bool])] = []
  conns: List[(Int, ServerConn)] = []
  next_conn: Int = 1
  next_lock: Int = 1
  slow_mgmt_ms: Int = 0
  gen: Int = 0
  detaches: List[(Int, Str, amqp.WireError)] = []
  closes: List[(Int, amqp.WireError?)] = []
}

struct Broker {
  st: Mutex[BrokerState]
  clock: Clock
}

fn (b: Broker) serve(g: task.Group, l: Listener) uses (net.listen, clock)

fn (b: Broker) queue(name: Str, opts: QueueOptions = QueueOptions{})

fn (b: Broker) event_hub(name: Str, partitions: Int, consumer_groups: List[Str] = ["$Default"])

fn (b: Broker) event_hub_geo_factor(hub: Str, factor: Int)

fn (b: Broker) events(hub: Str, partition: Int) -> List[amqp.Message]

fn (b: Broker) topic(name: Str)

fn (b: Broker) subscription(topic: Str, name: Str, opts: QueueOptions = QueueOptions{})

fn (b: Broker) put(queue: Str, m: amqp.Message)

fn (b: Broker) ready(queue: Str) -> List[amqp.Message]

fn (b: Broker) locked(queue: Str) -> Int

fn (b: Broker) dead_letters(queue: Str) -> List[amqp.Message]

fn (b: Broker) scheduled(queue: Str) -> List[amqp.Message]

fn (b: Broker) deferred(queue: Str) -> List[amqp.Message]

fn (b: Broker) tokens() -> List[(Str, Str)]

fn (b: Broker) attaches() -> List[(Str, Str, List[Str])]

fn (b: Broker) reject_token(token: Str)

fn (b: Broker) require_token(on: Bool)

fn (b: Broker) session_state(queue: Str, session: Str) -> Bytes

fn (b: Broker) fail_next(op: Str, status: Int, condition: Str, description: Str)

fn (b: Broker) fail(op: Str, status: Int, condition: Str, description: Str)

fn (b: Broker) clear_failures()

fn (b: Broker) slow_management(ms: Int)

fn (b: Broker) detach_links(queue: Str, error: amqp.WireError)

fn (b: Broker) disconnect_all(error: amqp.WireError? = None)

struct SvLink {
  handle: Int
  remote: Int
  channel: Int
  name: Str
  sends: Bool
  kind: Int
  queue: Str
  credit: Int = 0
  delivery_count: Int = 0
  presettled: Bool = false
  session_id: Str = ""
  partial: Bytes = b""
  partial_id: Int = 0
  partial_settled: Bool = false
  partial_active: Bool = false
  partial_format: Int = 0
  replies: List[amqp.Message] = []
  unsettled: List[(Int, Int)] = []
  detached: Bool = false
  hub: Str = ""
  cg: Str = ""
  part: Int = -1
  pos: Int = 0
  epoch: Int = 0
  hkey: Str = ""
}

struct SvSession {
  channel: Int
  next_in: Int = 0
  next_out: Int = 0
  remote_window: Int = 0
  ended: Bool = false
  in_used: Int = 0
}

struct SvConn {
  id: Int
  max_frame: Int = 65536
  sess: List[SvSession] = []
  links: List[SvLink] = []
  seen_gen: Int = 0
}

fn new(clock: Clock) -> Broker