Open pshashk opened 4 years ago
On the producer side, it seems like a simple state check is enough.
create_connection() = connection(;virtualhost="/", host=HOST, port=PORT, auth_params=AUTH)
create_channel(conn) = channel(conn, AMQPClient.UNUSED_CHANNEL, true)
mutable struct Producer
conn
chnl
function Producer()
conn = create_connection()
chnl = create_channel(conn)
success = exchange_declare(chnl, EXCHANGE, EXCHANGE_TYPE_FANOUT)
return new(conn, chnl)
end
end
function (p::Producer)()
if p.conn.state != AMQPClient.CONN_STATE_OPEN
p.conn = create_connection()
p.chnl = create_channel(p.conn)
elseif p.chnl.state != AMQPClient.CONN_STATE_OPEN
p.chnl = create_channel(p.conn)
end
timestamp = round(Int, time())
msg = Message(Vector{UInt8}(string(timestamp)), content_type="text/plain")
basic_publish(p.chnl, msg; exchange=EXCHANGE, routing_key="")
@info "sent" data=timestamp
end
But messages still don't make it to the consumer. So some connection handling is required there as well.
I've managed to implement consumer that could handle broker restarts. But I doubt that's the best way to do it.
function consume()
while true
try
conn = connection()
chnl = channel(conn, AMQPClient.UNUSED_CHANNEL, true)
success = exchange_declare(chnl, EXCHNG, EXCHANGE_TYPE_FANOUT, durable=true)
success, queue_name, _, _ = queue_declare(
chnl, ""; exclusive=true, durable=false, auto_delete=true
)
success = queue_bind(chnl, queue_name, EXCHNG, ROUTE)
callback = (msg) -> begin
@info "recieved" data=String(msg.data)
basic_ack(chnl, msg.delivery_tag)
end
success, consumer_tag = basic_consume(chnl, queue_name, callback)
fetch(chnl.consumers[consumer_tag].receiver)
catch e
@error exception=(e, stacktrace(catch_backtrace()))
sleep(5)
end
end
end
@pshashk This is excellent, thanks. My consumer service now restarts gracefully on connection error.
Communication via fanout exchange works fine.
But when I restart RabbitMQ (
docker restart ...
) and callproducer
again julia process hangs and eats all available memory without any error reporting.The RabbitMQ reliability guide says that in such cases it is necessary to create new connection and channel. How can I do that with AMQPClient?