Bind every key a node may publish
The control queue was bound to enrol and not to report, so every report a node sent was accepted by the broker, matched no binding, and dropped. The publisher saw success and the consumer saw nothing, for an afternoon. The refactor that was meant to bind both never applied -- it left behind a helper nothing called, which compiled and passed vet. The loop is now where the bind is, so there is one place to forget rather than two.
This commit is contained in:
+9
-11
@@ -68,23 +68,21 @@ func Connect(enroller Enroller) (*Server, error) {
|
||||
conn.Close()
|
||||
return nil, fmt.Errorf("cannot declare the %s queue: %w", ControlQueue, err)
|
||||
}
|
||||
if err := channel.QueueBind(ControlQueue, KeyEnrol, Exchange, false, nil); err != nil {
|
||||
conn.Close()
|
||||
return nil, err
|
||||
// Every key a node may publish. Binding one and forgetting another is a message the broker
|
||||
// accepts, finds no queue for, and drops — the publisher sees success and the consumer sees
|
||||
// nothing. That is exactly what happened to reports: `report` was left unbound while `enrol`
|
||||
// worked, so nodes announced what they had applied into a void for an afternoon.
|
||||
for _, key := range []string{KeyEnrol, KeyReport} {
|
||||
if err := channel.QueueBind(ControlQueue, key, Exchange, false, nil); err != nil {
|
||||
conn.Close()
|
||||
return nil, fmt.Errorf("cannot bind %s to %s/%s: %w", ControlQueue, Exchange, key, err)
|
||||
}
|
||||
}
|
||||
|
||||
return &Server{conn: conn, channel: channel, enroller: enroller,
|
||||
log: log.New(os.Stdout, "", log.LstdFlags)}, nil
|
||||
}
|
||||
|
||||
func (s *Server) bindOrClose(channel *amqp.Channel, conn *amqp.Connection, key string) error {
|
||||
if err := channel.QueueBind(ControlQueue, key, Exchange, false, nil); err != nil {
|
||||
conn.Close()
|
||||
return fmt.Errorf("cannot bind %s to %s/%s: %w", ControlQueue, Exchange, key, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Channel is the control plane's channel, for sending declarations.
|
||||
func (s *Server) Channel() *amqp.Channel { return s.channel }
|
||||
|
||||
|
||||
Reference in New Issue
Block a user