DeepMQ — Low-Level Design Architecture ┌─────────────────────────────────────────────────────────────────────────────────┐
│ DeepMQBroker │
│ (src/server.ts) │
│ │
│ ┌─────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────────┐ │
│ │ TCP Server │ │ Connection │ │ Channel │ │ BrokerEvent │ │
│ │ (net.Server)│──▶│ Manager │──▶│ Manager │ │ Emitter │ │
│ └─────────────┘ └──────────────┘ └──────────────┘ └──────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────────────────────┐│
│ │ Protocol Layer ││
│ │ ┌──────────────┐ ┌──────────────┐ ┌───────────────────────────────────┐ ││
│ │ │ FrameParser │ │ FrameBuilder │ │ Method Handlers │ ││
│ │ │ (TCP→Frames) │ │ (Frames→TCP) │ │ connection│channel│exchange│ │ ││
│ │ │ │ │ │ │ queue│basic │ ││
│ │ └──────────────┘ └──────────────┘ └───────────────────────────────────┘ ││
│ └─────────────────────────────────────────────────────────────────────────────┘│
│ │
│ ┌─────────────────────────────────────────────────────────────────────────────┐│
│ │ Core Layer ││
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌─────────────┐ ││
│ │ │ Exchange │ │ Queue │ │ Binding │ │ Consumer │ │ Message │ ││
│ │ │ │ │ │ │ Registry │ │ Registry │ │ │ ││
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ └─────────────┘ ││
│ └─────────────────────────────────────────────────────────────────────────────┘│
│ │
│ ┌─────────────────────────────┐ ┌────────────────────────────────────────────┐│
│ │ Routing Layer │ │ Persistence Layer ││
│ │ ┌───────────────────────┐ │ │ ┌────────────────────────────────────┐ ││
│ │ │ MessageRouter │ │ │ │ FileStore │ ││
│ │ │ ┌─────────────────┐ │ │ │ │ messages.log (append-only) │ ││
│ │ │ │ DirectExchange │ │ │ │ │ queues.json (atomic write) │ ││
│ │ │ │ FanoutExchange │ │ │ │ │ exchanges.json (atomic write) │ ││
│ │ │ │ TopicExchange │ │ │ │ │ bindings.json (atomic write) │ ││
│ │ │ └─────────────────┘ │ │ │ └────────────────────────────────────┘ ││
│ │ └───────────────────────┘ │ │ ┌────────────────────────────────────┐ ││
│ └─────────────────────────────┘ │ │ Recovery Module │ ││
│ │ │ (load on startup) │ ││
│ │ └────────────────────────────────────┘ ││
│ └────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────────────────────────────┘
Component
File
Responsibility
TCP Server server.tsAccepts TCP connections on port 5672
Connection core/connection.tsPer-client TCP socket state machine, heartbeat management
Channel core/channel.tsMultiplexed logical channel within a connection
Component
File
Responsibility
FrameParser protocol/frame-parser.tsTCP byte stream → typed AMQP frames
FrameBuilder protocol/frame-builder.tsTyped data → AMQP wire format frames
Method Handlers protocol/methods/*.tsParse/build class-specific AMQP methods
Component
File
Responsibility
Exchange core/exchange.tsNamed routing entity (direct, fanout, topic, headers)
Queue core/queue.tsIn-memory message buffer with consumer tracking
Binding core/binding.tsExchange→Queue routing rule with key and registry
Consumer core/consumer.tsSubscription handle with registry & queue index
Message core/message.tsContent envelope: body, properties, routing metadata
Component
File
Responsibility
MessageRouter routing/router.tsDispatches messages through exchange→binding→queue chain
DirectExchange routing/direct-exchange.tsExact routing key match
FanoutExchange routing/fanout-exchange.tsBroadcast to all bound queues
TopicExchange routing/topic-exchange.tsWildcard pattern matching (*, #)
Component
File
Responsibility
FileStore persistence/file-store.tsAppend-only log + atomic JSON files for durability
Recovery persistence/recovery.tsRebuild broker state from disk on startup
Component
File
Responsibility
BrokerEventEmitter events/broker-events.tsTyped event bus for all broker lifecycle events
TCP connect
│
▼
┌──────────────────┐ AMQP\x00\x00\x09\x01
│ awaiting_header │ ────────────────────────┐
└──────────────────┘ │
▼
┌──────────────────────┐
┌────────│ awaiting_start_ok │
│ └──────────────────────┘
│ Connection.Start-Ok (credentials)
▼
┌──────────────────────┐
│ awaiting_tune_ok │
└──────────────────────┘
│ Connection.Tune-Ok
▼
┌──────────────────────┐
│ awaiting_open │
└──────────────────────┘
│ Connection.Open (vhost)
▼
┌──────────────────────┐
│ open │◄─── heartbeat loop
└──────────────────────┘
│ Connection.Close
▼
┌──────────────────────┐
│ closing │
└──────────────────────┘
│ Connection.Close-Ok
▼
┌──────────────────────┐
│ closed │ ──▶ socket.destroy()
└──────────────────────┘
Producer (AMQP Client) DeepMQ Broker
═══════════════════════ ══════════════
┌─────────────────┐
│ Basic.Publish │──── Method Frame ────▶ handleMethodFrame()
│ (exchange, key) │ store in channel.pendingMessage
└─────────────────┘
┌─────────────────┐
│ Content Header │──── Header Frame ────▶ handleHeaderFrame()
│ (size, props) │ set expectedBodySize, properties
└─────────────────┘
┌─────────────────┐
│ Content Body │──── Body Frame(s) ───▶ handleBodyFrame()
│ (payload chunks) │ accumulate until bodySize reached
└─────────────────┘
│
▼
completeMessage()
│
┌────────────────┼────────────────┐
▼ ▼ ▼
emit event MessageRouter Persist if
'message:published' .route() durable + mode=2
│
┌──────────────┼──────────────┐
▼ ▼ ▼
Exchange Binding Queue(s)
lookup match .enqueue()
│ │ │
│ │ ▼
│ │ deliverToConsumer()
│ │ (setImmediate loop)
│ │
▼ │
No routes + mandatory? │
│ │
┌───┴────┐ │
│ Yes │ │
▼ │ │
Basic.Return │ │
to publisher │ │
│ │
▼ ▼
emit 'message:routed'
Consumer (AMQP Client) DeepMQ Broker
═══════════════════════ ══════════════
┌──────────────────┐
│ Basic.Consume │────────────────────▶ Register Consumer
│ (queue, tag, ack) │ ConsumerRegistry.add()
└──────────────────┘ Channel.addConsumer()
│
▼
deliverToConsumer()
│
┌──────────┴──────────┐
│ QoS check: │
│ unacked < prefetch? │
└──────────┬──────────┘
yes │ no
┌─────┘ └─── wait for ack
▼
queue.dequeue()
│
▼
channel.nextDeliveryTag()
channel.trackUnacked()
│
┌──────────────────┐ │
│ Basic.Deliver │◀── Method Frame ───────┘
│ Content Header │◀── Header Frame
│ Content Body │◀── Body Frame(s)
└──────────────────┘
│
│ (client processes message)
▼
┌──────────────────┐
│ Basic.Ack │────────────────────▶ channel.ack(deliveryTag)
│ (deliveryTag) │ │
└──────────────────┘ ┌────────┼────────┐
▼ ▼ ▼
remove persist deliver
unacked .delete() next msg
to consumer
─── OR ───
┌──────────────────┐
│ Basic.Nack/Reject │────────────────────▶ channel.nack(tag, requeue)
└──────────────────┘ │
┌────────┴────────┐
▼ ▼
requeue=true requeue=false
queue.requeue() delete message
redeliver
Exchange Routing Strategies ┌──────────────┐
│ Exchange │
│ (source) │
└──────┬───────┘
│
┌────────────┼────────────┐
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ direct │ │ fanout │ │ topic │
└────┬─────┘ └────┬─────┘ └────┬─────┘
│ │ │
▼ ▼ ▼
routingKey ignore key pattern match
== binding return ALL * = one word
key (exact) bound queues # = zero+ words
│ │ │
▼ ▼ ▼
┌───────────┐ ┌────────────┐ ┌─────────────────────────┐
│ Queue A │ │ Queue A │ │ "stock.#" → Queue A │
│ (key=rk1) │ │ Queue B │ │ "*.error" → Queue B │
└───────────┘ │ Queue C │ │ "log.#.warn" → Queue C │
└────────────┘ └─────────────────────────┘
Default Exchange (name=""):
Implicit direct routing: routingKey == queueName
Every queue auto-bound on creation
┌──────────────────┐
│ FileStore │
│ (data/ dir) │
└────────┬─────────┘
│
┌────────────────┼──────────────────┐
▼ ▼ ▼
┌─────────────┐ ┌────────────┐ ┌──────────────┐
│ messages.log │ │ Metadata │ │ Recovery │
│ (append-only)│ │ JSON files │ │ Module │
└──────┬──────┘ └─────┬──────┘ └──────┬───────┘
│ │ │
▼ ▼ ▼
┌─────────────┐ ┌────────────┐ On startup:
│ JSON Lines: │ │queues.json │ 1. Load exchanges
│ {type:msg, │ │exchanges. │ 2. Load queues
│ queue, │ │ json │ 3. Replay messages.log
│ data:b64, │ │bindings. │ 4. Load bindings
│ checksum} │ │ json │ 5. Create defaults
│ │ └────────────┘
│ {type:del, │ │
│ queue, │ ▼
│ messageId} │ Atomic write:
└──────┬──────┘ .tmp → rename
│
▼
Compaction:
Rewrites log with
only live messages
(removes deletes)
Frame Processing Pipeline TCP Socket FrameParser Broker Handler
══════════ ═══════════ ══════════════
raw bytes ──▶ append(data)
│
▼
┌───────────┐
│ Buffer │◀── accumulates until
│ (partial) │ complete frame
└─────┬─────┘
│ hasCompleteFrame()
▼
parseFrame()
┌─────────────────────────────────┐
│ [type:1B][channel:2B][size:4B] │
│ [payload: size bytes] │
│ [frame_end: 0xCE] │
└─────────────────────────────────┘
│
┌─────────┼──────────┬─────────────┐
▼ ▼ ▼ ▼
type=1 type=2 type=3 type=8
METHOD HEADER BODY HEARTBEAT
│ │ │ │
▼ ▼ ▼ ▼
parseMethod parseCH raw Buffer resetHeartbeat
(classId, (bodySize, send heartbeat
methodId, props) back
args) │ │
│ └────┬─────┘
│ ▼
│ channel.pendingMessage
│ (accumulate multi-frame)
│ │
▼ ▼
Route by classId: When bodySize reached:
10=connection completeMessage()
20=channel │
40=exchange ▼
50=queue route + deliver
60=basic
Channel
┌──────────────────┐
│ prefetchCount: 5 │
│ unacked: [■■■□□] │ (3 of 5 used)
└────────┬─────────┘
│
canDeliver()? │
┌──────────────┤
▼ ▼
unacked.size prefetchCount
(3) < (5)
│
▼
true → deliver next message
trackUnacked(msg)
unacked: [■■■■□] (4 of 5)
...after Basic.Ack...
unacked: [■■■□□] (3 of 5)
→ deliver another message
Broker Action Event Emitted
═════════════ ═════════════
TCP accept ──▶ connection:open
Auth failure / close ──▶ connection:close
Channel.Open-Ok ──▶ channel:open
Exchange.Declare (new) ──▶ exchange:created
Queue.Declare (new) ──▶ queue:created
Queue.Bind ──▶ binding:created
Basic.Consume ──▶ consumer:created
Basic.Publish (assembled) ──▶ message:published
Router matched queues ──▶ message:routed
Basic.Deliver sent ──▶ message:delivered
Basic.Ack received ──▶ message:acked
Basic.Nack received ──▶ message:nacked
Basic.Reject received ──▶ message:rejected
No route + mandatory ──▶ message:returned
Queue.Delete ──▶ queue:deleted
broker.start() ──▶ broker:started
broker.stop() ──▶ broker:stopped
Startup / Shutdown Sequence START STOP
═════ ════
FileStore.initialize() Close all connections
│ (requeue unacked msgs)
▼ │
Recovery.recoverFromStore() ▼
├─ load durable exchanges Close TCP server
├─ load durable queues │
├─ replay messages.log ▼
└─ load bindings FileStore.close()
│ │
▼ ▼
createDefaultExchanges() emit 'broker:stopped'
├─ "" (default direct)
├─ amq.direct
├─ amq.fanout
└─ amq.topic
│
▼
Update router context
│
▼
TCP server.listen(5672)
│
▼
emit 'broker:started'
Data Structures & Indexes Broker State (in-memory)
════════════════════════
connections: Map<connId, Connection>
│
└──▶ Connection.channels: Map<channelNum, Channel>
│
└──▶ Channel.consumers: Map<tag, ConsumerDef>
└──▶ Channel.unackedMessages: Map<deliveryTag, UnackedMessage>
exchanges: Map<name, Exchange>
queues: Map<name, Queue>
│
└──▶ Queue.messages: Message[] (in-memory FIFO)
bindings: BindingRegistry
├── bindings: Map<key, Binding> key = "src:dst:routingKey"
├── bySource: Map<exchange, Set<key>> fast "all from exchange"
└── byDestination: Map<queue, Set<key>> fast "all to queue"
consumers: ConsumerRegistry
├── consumers: Map<tag, Consumer>
└── queueConsumers: Map<queue, Set<tag>> fast "all on queue"
router: MessageRouter
└── context: { exchanges, queues, bindings }
store: FileStore
└── messagesByQueue: Map<queue, Map<msgId, Message>>