{"action":"created","issue":{"url":"https://api.github.com/repos/andersfugmann/amqp-client/issues/30","repository_url":"https://api.github.com/repos/andersfugmann/amqp-client","labels_url":"https://api.github.com/repos/andersfugmann/amqp-client/issues/30/labels{/name}","comments_url":"https://api.github.com/repos/andersfugmann/amqp-client/issues/30/comments","events_url":"https://api.github.com/repos/andersfugmann/amqp-client/issues/30/events","html_url":"https://github.com/andersfugmann/amqp-client/issues/30","id":376337071,"node_id":"MDU6SXNzdWUzNzYzMzcwNzE=","number":30,"title":"Doesn't catch the on_closed \"event\"","user":{"login":"XVilka","id":203261,"node_id":"MDQ6VXNlcjIwMzI2MQ==","avatar_url":"https://avatars1.githubusercontent.com/u/203261?v=4","gravatar_id":"","url":"https://api.github.com/users/XVilka","html_url":"https://github.com/XVilka","followers_url":"https://api.github.com/users/XVilka/followers","following_url":"https://api.github.com/users/XVilka/following{/other_user}","gists_url":"https://api.github.com/users/XVilka/gists{/gist_id}","starred_url":"https://api.github.com/users/XVilka/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/XVilka/subscriptions","organizations_url":"https://api.github.com/users/XVilka/orgs","repos_url":"https://api.github.com/users/XVilka/repos","events_url":"https://api.github.com/users/XVilka/events{/privacy}","received_events_url":"https://api.github.com/users/XVilka/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":7,"created_at":"2018-11-01T10:32:50Z","updated_at":"2018-11-05T21:01:37Z","closed_at":null,"author_association":"NONE","body":"Sorry for the long paste, but it is vital to show how reconnect is implemented.\r\nSo basically, if `info_query` function takes a long time and CPU/memory resourses, sometimes RabbitMQ connection (or channel) get dropped. As you implemented `on_closed` recently, I added a reconnection handler for this \"event\", but it seems never called again - when heavy-load query is processed and `info_query` finishes its job, the connection never reinstates, and all messages waiting in the queue are never received. Even if I set a timeout for messages and resend them from producer again, this worker never get them anyway. So it seems the connection was never properly resurrected.\r\n\r\n```ocaml\r\ntype handler = Amqp_client_lwt.Message.t -> unit Lwt.t\r\ntype reader = Amqp_client_lwt.Message.t Amqp_client_lwt.Thread.Pipe.Reader.t\r\ntype consumer = Amqp_client_lwt.Channel.no_confirm Amqp_client_lwt.Queue.consumer\r\n\r\ntype queue = {\r\n    name: string;\r\n    topic: string;\r\n    queue: Amqp_client_lwt.Queue.t;\r\n    mutable handler: handler option;\r\n    mutable consumer: consumer option;\r\n    mutable reader: reader option;\r\n}\r\n\r\n(* Type to preserve the connection state *)\r\ntype conn = {\r\n    mutable connection: Amqp_client_lwt.Connection.t option;\r\n    mutable channel: Amqp_client_lwt.Channel.no_confirm Amqp_client_lwt.Channel.t option;\r\n    mutable exchange: [`Topic of string] Amqp_client_lwt.Exchange.t option;\r\n    mutable control_queue: queue option;\r\n    mutable errors_queue: queue option;\r\n    mutable queues: queue list;\r\n}\r\n\r\n(* Global connection, channel, exchange, and bindings state *)\r\nlet conn = {\r\n    connection = None;\r\n    channel = None;\r\n    exchange = None;\r\n    control_queue = None;\r\n    errors_queue = None;\r\n    queues = [];\r\n}\r\n\r\n(* Internal helper functions *)\r\nmodule Internal = struct\r\n\r\n    (* Creates an unresolved connection instance *)\r\n    let make_connection () =\r\n        let open Amqp in\r\n        match Conf.AMQP.amqp_port with\r\n        | None ->\r\n            Connection.connect ~id:connection_id\r\n                ~credentials:(Conf.AMQP.amqp_user, Conf.AMQP.amqp_pass)\r\n                ~virtual_host:Conf.AMQP.amqp_vhost Conf.AMQP.amqp_host\r\n        | Some p ->\r\n            Connection.connect ~id:connection_id ~port:p\r\n                ~credentials:(Conf.AMQP.amqp_user, Conf.AMQP.amqp_pass)\r\n                ~virtual_host:Conf.AMQP.amqp_vhost Conf.AMQP.amqp_host\r\n\r\n    (* Creates an unresolved channel instance *)\r\n    let make_channel connection =\r\n        Connection.open_channel ~id:channel_id\r\n            Channel.no_confirm connection\r\n\r\n    (* Creates an unresolved exchange instance *)\r\n    let make_exchange channel =\r\n        Exchange.declare ~durable:true channel\r\n            Exchange.topic_t Conf.AMQP.exchange\r\n\r\n    (* Creates an unresolved queue binding *)\r\n    let make_queue channel exchange ~name ~topic =\r\n        Queue.declare channel name >>= fun queue ->\r\n        Queue.bind channel queue exchange (`Topic topic) >>= fun () ->\r\n        Lwt.return queue\r\n\r\n    (* Creates unresolved standard queues *)\r\n    let make_standard_queues channel exchange =\r\n        (* Special Control queue for sending control messages *)\r\n        make_queue channel exchange\r\n            ~name:control_queue_name ~topic:control_queue_topic\r\n        >>= fun control_queue ->\r\n        make_queue channel exchange\r\n            ~name:errors_queue_name ~topic:errors_queue_topic\r\n        >>= fun errors_queue ->\r\n        Lwt.return (control_queue, errors_queue)\r\n\r\n    (* Creates an unresolved (consumer, reader) pair *)\r\n    let make_queue_consumer ~(worker:worker_record) channel queue handler =\r\n        let consumer_cancelled () =\r\n            Lwt_io.printf \"Cancelled consuming for queue.\"; ()\r\n        in\r\n        Queue.consume ~no_ack:true ~id:worker.name\r\n            ~on_cancel:consumer_cancelled channel queue\r\n\r\nend\r\n\r\n(* ----------------------------------------------------------------------------- *)\r\n\r\n(* Recreate the connection from scratch *)\r\nlet rec recreate_connection ~worker conn =\r\n    Internal.make_connection ()\r\n    >>= fun connection ->\r\n    (* Set the reconnect handler for 'on_closed' *)\r\n    Thread.spawn (Connection.on_closed connection >>= fun () ->\r\n        connection_on_closed_handler ~worker\r\n    );\r\n    Lwt.return connection\r\n\r\n(* Handle the case of closed connection, restarting sequence *)\r\nand reconnect ~worker conn =\r\n    recreate_connection ~worker conn >>= fun connection ->\r\n    (* Override the old connection instance with a new one *)\r\n    conn.connection <- Some connection;\r\n    Lwt.return_unit\r\n\r\nand connection_on_closed_handler ~worker =\r\n    Lwt_io.printf \"Connection closed, reconnecting...\\n\" >>= fun () ->\r\n    (* Restart connecting sequence *)\r\n    reconnect ~worker conn >>= fun () ->\r\n    (* Restart opening channel sequence *)\r\n    reopen_channel ~worker conn >>= fun () ->\r\n    (* Recreate exchange *)\r\n    redefine_exchange ~worker conn >>= fun () ->\r\n    (* Recreate standard queues and their bindings *)\r\n    redefine_standard_queues ~worker conn >>= fun () ->\r\n    (* Recreate other queues and their bindings *)\r\n    redefine_other_queues ~worker conn >>= fun () ->\r\n    (* Restart all the consumers *)\r\n    restart_consumers ~worker conn >>= fun () ->\r\n    Lwt.return_unit\r\n\r\n(* ----------------------------------------------------------------------------- *)\r\n\r\n(* Recreate the channel from scratch *)\r\nand recreate_channel ~worker conn =\r\n    match conn.connection with\r\n    | Some connection ->\r\n            Internal.make_channel connection\r\n    | None -> (\r\n            recreate_connection ~worker conn >>= fun connection ->\r\n            conn.connection <- Some connection;\r\n            Internal.make_channel connection\r\n    )\r\n    >>= fun channel ->\r\n    (* Set the reopen handler for 'on_closed' *)\r\n    Thread.spawn (Channel.on_closed channel >>= fun () ->\r\n        channel_on_closed_handler ~worker\r\n    );\r\n    Lwt.return channel\r\n\r\n(* Handle the case of closed exchange, restarting sequence *)\r\nand reopen_channel ~worker conn =\r\n    recreate_channel ~worker conn >>= fun channel ->\r\n    (* Override the old channel instance with a new one *)\r\n    conn.channel <- Some channel;\r\n    Lwt.return_unit\r\n\r\nand channel_on_closed_handler ~worker =\r\n    Lwt_io.printf \"Channel closed, reopening...\\n\" >>= fun () ->\r\n    (* Restart opening channel sequence *)\r\n    reopen_channel ~worker conn >>= fun () ->\r\n    (* Recreate exchange *)\r\n    redefine_exchange ~worker conn >>= fun () ->\r\n    (* Recreate standard queues and their bindings *)\r\n    redefine_standard_queues ~worker conn >>= fun () ->\r\n    (* Recreate other queues and their bindings *)\r\n    redefine_other_queues ~worker conn >>= fun () ->\r\n    (* Restart all the consumers *)\r\n    restart_consumers ~worker conn >>= fun () ->\r\n    Lwt.return_unit\r\n\r\n(* Recreate the exchange from scratch *)\r\nand recreate_exchange ~(worker:worker_record) conn =\r\n    match conn.channel with\r\n    | Some channel -> (\r\n        (* Make a new exchange instance *)\r\n        Internal.make_exchange channel >>= (fun exchange ->\r\n        (* Override the old exchange instance with a new one *)\r\n        conn.exchange <- Some exchange;\r\n        Lwt.return exchange)\r\n    )\r\n    | None -> (\r\n        recreate_channel ~worker conn >>= (fun channel ->\r\n        (* Override the old channel instance with a new one *)\r\n        conn.channel <- Some channel;\r\n        (* Make a new exchange instance *)\r\n        Internal.make_exchange channel >>= (fun exchange ->\r\n        (* Override the old exchange instance with a new one *)\r\n        conn.exchange <- Some exchange;\r\n        Lwt.return exchange))\r\n    )\r\n\r\n(* Make a new exchange instance *)\r\nand redefine_exchange ~worker conn =\r\n    recreate_exchange ~worker conn >>= fun exchange ->\r\n    (* Override the old exchange instance with a new one *)\r\n    conn.exchange <- Some exchange;\r\n    Lwt.return_unit\r\n\r\n(* Recreate standard queues instances from scratch *)\r\nand recreate_standard_queues ~worker conn =\r\n    match conn.channel with\r\n    | Some channel -> (\r\n        match conn.exchange with\r\n        | Some exchange -> (\r\n            (* Create special control and errors queues *)\r\n            Internal.make_standard_queues channel exchange)\r\n        | None -> (\r\n            recreate_exchange ~worker conn >>= (fun exchange ->\r\n            (* Override the old exchange instance with a new one *)\r\n            conn.exchange <- Some exchange;\r\n            (* Create special control and errors queues *)\r\n            Internal.make_standard_queues channel exchange)\r\n        )\r\n    )\r\n    | None -> (\r\n        recreate_channel ~worker conn >>= (fun channel ->\r\n        (* Override the old channel instance with a new one *)\r\n        conn.channel <- Some channel;\r\n        (* Make a new exchange instance *)\r\n        Internal.make_exchange channel >>= (fun exchange ->\r\n        (* Create special control and errors queues *)\r\n        Internal.make_standard_queues channel exchange))\r\n    )\r\n\r\n(* Remake a new exchange instance *)\r\nand redefine_standard_queues ~worker conn =\r\n    (* Check if the channel, connection and exchanges are fine before\r\n     * the recreating all queues and bindings *)\r\n    recreate_standard_queues ~worker conn\r\n    >>= fun (control_queue, errors_queue) ->\r\n    conn.control_queue <- Some {\r\n            name = control_queue_name;\r\n            topic = control_queue_topic;\r\n            queue = control_queue;\r\n            handler = None;\r\n            consumer = None;\r\n            reader = None;\r\n        };\r\n    conn.errors_queue <- Some {\r\n            name = errors_queue_name;\r\n            topic = errors_queue_topic;\r\n            queue = errors_queue;\r\n            handler = None;\r\n            consumer = None;\r\n            reader = None;\r\n        };\r\n    Lwt.return_unit\r\n\r\n\r\nand recreate_other_queues ~(worker:worker_record) channel exchange queues =\r\n    Lwt_list.map_s (fun queue ->\r\n        Lwt_io.printf \"Rebinding the queue \\\"%s\\\" for topic \\\"%s\\\"\\n\"\r\n            queue.name queue.topic >>= fun () ->\r\n        Internal.make_queue channel exchange ~name:queue.name ~topic:queue.topic\r\n        >>= fun newqueue ->\r\n        (* Rebind the handler if exists *)\r\n        match queue.handler with\r\n        | Some handler ->\r\n            Internal.make_queue_consumer ~worker channel newqueue handler\r\n            >>= fun (consumer, reader) ->\r\n            Lwt.return {\r\n                name = queue.name;\r\n                topic = queue.topic;\r\n                queue = newqueue;\r\n                handler = queue.handler;\r\n                consumer = Some consumer;\r\n                reader = Some reader;\r\n            }\r\n        | None ->\r\n            Lwt.return { queue with queue = newqueue; }\r\n    ) queues\r\n\r\n(* Remake the rest attached queues *)\r\nand redefine_other_queues ~worker conn =\r\n    (* Check if the channel, connection and exchanges are fine before\r\n     * the recreating all queues and bindings *)\r\n    match conn.channel with\r\n    | Some channel -> (\r\n        match conn.exchange with\r\n        | Some exchange -> (\r\n            recreate_other_queues ~worker channel exchange conn.queues\r\n            >>= fun newqueues ->\r\n            conn.queues <- newqueues;\r\n            Lwt.return_unit\r\n        )\r\n        | None -> (\r\n            recreate_exchange ~worker conn >>= fun exchange ->\r\n            (* Override the old exchange instance with a new one *)\r\n            conn.exchange <- Some exchange;\r\n            Lwt.return_unit\r\n        )\r\n    )\r\n    | None -> (\r\n        recreate_channel ~worker conn >>= fun channel ->\r\n        (* Override the old channel instance with a new one *)\r\n        conn.channel <- Some channel;\r\n        (* Make a new exchange instance *)\r\n        Internal.make_exchange channel >>= fun exchange ->\r\n        (* Override the old exchange instance with a new one *)\r\n        conn.exchange <- Some exchange;\r\n        (* Create the queues *)\r\n        recreate_other_queues ~worker channel exchange conn.queues\r\n        >>= fun newqueues ->\r\n        conn.queues <- newqueues;\r\n        Lwt.return_unit\r\n    )\r\n\r\nand restart_consumers ~worker conn =\r\n    let process_messages () =\r\n        Lwt.join (\r\n            (* At first - restart the consumering for standard queues *)\r\n            let standard_queues =\r\n                match conn.control_queue, conn.errors_queue with\r\n                | Some control_queue, Some errors_queue ->\r\n                    [control_queue; errors_queue]\r\n                | _, _ -> [] (* Error? *)\r\n            in\r\n            (* Then for the rest of them *)\r\n            List.map ~f:(fun queue ->\r\n                match queue.handler, queue.consumer, queue.reader with\r\n                | Some handler, Some consumer, Some reader ->\r\n                    Lwt_io.printf \"Restarting consumering for \\\"%s\\\" (\\\"%s\\\") queue...\\n\"\r\n                        queue.name queue.topic >>= fun () ->\r\n                    Thread.Pipe.iter_without_pushback reader ~f:(fun m ->\r\n                        handler m; ()\r\n                    )\r\n                | _, _, _ -> Lwt.return_unit\r\n            ) (List.append standard_queues conn.queues)\r\n        )\r\n    in\r\n    Lwt_result.catch (process_messages ()) >>= fun _ ->\r\n    (* Cancel all consumers here *)\r\n    Lwt_list.iter_s (fun queue ->\r\n        match queue.consumer with\r\n        | Some consumer -> Queue.cancel consumer\r\n        | None -> Lwt.return_unit\r\n    ) conn.queues >>= fun () ->\r\n    Lwt.return_unit\r\n\r\n(* ----------------------------------------------------------------------------- *)\r\n\r\n(* Returns initialized connection, channel, exchange *)\r\n(* Returns also some predefined queues *)\r\nlet connect ~worker =\r\n    reconnect ~worker conn >>= fun () ->\r\n    reopen_channel ~worker conn >>= fun () ->\r\n    redefine_exchange ~worker conn >>= fun () ->\r\n    redefine_standard_queues ~worker conn >>= fun () ->\r\n    Lwt.return conn\r\n\r\nlet make_queue ~worker conn ~name ~topic =\r\n    match conn.channel with\r\n    | Some channel -> (\r\n        match conn.exchange with\r\n        | Some exchange -> (\r\n            Internal.make_queue channel exchange ~name ~topic >>= fun queue ->\r\n            Lwt.return {\r\n                name;\r\n                topic;\r\n                queue;\r\n                handler = None;\r\n                consumer = None;\r\n                reader = None;\r\n            }\r\n        )\r\n        | None -> (\r\n            recreate_exchange ~worker conn >>= fun exchange ->\r\n            (* Override the old exchange instance with a new one *)\r\n            conn.exchange <- Some exchange;\r\n            Internal.make_queue channel exchange ~name ~topic >>= fun queue ->\r\n            Lwt.return {\r\n                name;\r\n                topic;\r\n                queue;\r\n                handler = None;\r\n                consumer = None;\r\n                reader = None;\r\n            }\r\n        )\r\n    )\r\n    | None -> (\r\n        recreate_channel ~worker conn >>= fun channel ->\r\n        (* Override the old channel instance with a new one *)\r\n        conn.channel <- Some channel;\r\n        (* Make a new exchange instance *)\r\n        Internal.make_exchange channel >>= fun exchange ->\r\n        (* Override the old exchange instance with a new one *)\r\n        conn.exchange <- Some exchange;\r\n        Internal.make_queue channel exchange ~name ~topic >>= fun queue ->\r\n        Lwt.return {\r\n            name;\r\n            topic;\r\n            queue;\r\n            handler = None;\r\n            consumer = None;\r\n            reader = None;\r\n        }\r\n    )\r\n\r\n(* Register a queue + consumer pair *)\r\nlet register_queue ~worker conn queue handler =\r\n    match conn.channel with\r\n    | Some channel -> (\r\n        Internal.make_queue_consumer ~worker channel queue.queue handler\r\n        >>= fun (consumer, reader) ->\r\n        Lwt.return { queue with\r\n            handler = Some handler;\r\n            consumer = Some consumer;\r\n            reader = Some reader;\r\n        }\r\n    )\r\n    | None -> (\r\n        recreate_channel ~worker conn >>= fun channel ->\r\n        (* Override the old channel instance with a new one *)\r\n        conn.channel <- Some channel;\r\n        (* Make a new exchange instance *)\r\n        recreate_exchange ~worker conn >>= fun exchange ->\r\n        (* Override the old exchange instance with a new one *)\r\n        conn.exchange <- Some exchange;\r\n        Internal.make_queue_consumer ~worker channel queue.queue handler\r\n        >>= fun (consumer, reader) ->\r\n        Lwt.return { queue with\r\n            handler = Some handler;\r\n            consumer = Some consumer;\r\n            reader = Some reader;\r\n        }\r\n    )\r\n\r\n(* --------------------------------------------------------------------------------------------- *)\r\n\r\n(* Return a handler for all QUERY packets *)\r\nlet query_queue_consumer ~(worker:Cluster.worker_record) conn queue =\r\n    (* Here we consume only query messages *)\r\n    let handler (msg:Message.t) =\r\n        let _content, body = msg.message in\r\n        Lwt_io.printf \"query queue msg handler\\n\";\r\n        let (msgkind, protbuf) = decode_msg body in\r\n        let result = match msgkind with\r\n        | Query_notif -> info_query ~worker conn queue protbuf\r\n        | Query_ack -> blacklist_query ~worker conn protbuf\r\n        | Query_answer -> unblacklist_query ~worker conn protbuf\r\n        | _ -> None\r\n        in\r\n        Lwt.return_unit\r\n    in\r\n    handler\r\n\r\nlet run_worker ~worker conn =\r\n    Lwt_io.printf \"Starting listening threads...\\n\" >>= fun () ->\r\n    (* Create queue for handling queries *)\r\n    Cluster.make_queue ~worker conn ~name:\"query.manager\" ~topic:\"query.#\"\r\n    >>= fun query_queue ->\r\n    (* Registering itself *)\r\n    Lwt_io.printf \"Registering worker...\\n\" >>= fun () ->\r\n    Cluster.regsend ~worker conn >>= fun () ->\r\n    (* Registering handlers *)\r\n    (* Check if control queue exists first! *)\r\n    (match conn.control_queue with\r\n    | Some control_queue -> (\r\n        let handler_control: Message.t -> unit Lwt.t =\r\n            control_queue_consumer ~worker conn control_queue in\r\n        Cluster.register_queue ~worker conn control_queue handler_control\r\n        >>= fun control_queue ->\r\n        (* Override control queue in the connection state *)\r\n        conn.control_queue <- Some control_queue;\r\n        let handler_query: Message.t -> unit Lwt.t =\r\n            query_queue_consumer ~worker conn query_queue in\r\n        Cluster.register_queue ~worker conn query_queue handler_query\r\n        >>= fun query_queue ->\r\n        (* Add the query queue to the connection state *)\r\n        conn.queues <- (List.append conn.queues [query_queue]);\r\n        Cluster.restart_consumers ~worker conn\r\n    )\r\n    | None -> (\r\n        Lwt_io.printf \"ERROR: No control queue initialized!\\n\"\r\n    ))\r\n    >>= fun () ->\r\n    Lwt.return_unit\r\n\r\n```"},"comment":{"url":"https://api.github.com/repos/andersfugmann/amqp-client/issues/comments/436033246","html_url":"https://github.com/andersfugmann/amqp-client/issues/30#issuecomment-436033246","issue_url":"https://api.github.com/repos/andersfugmann/amqp-client/issues/30","id":436033246,"node_id":"MDEyOklzc3VlQ29tbWVudDQzNjAzMzI0Ng==","user":{"login":"andersfugmann","id":647024,"node_id":"MDQ6VXNlcjY0NzAyNA==","avatar_url":"https://avatars2.githubusercontent.com/u/647024?v=4","gravatar_id":"","url":"https://api.github.com/users/andersfugmann","html_url":"https://github.com/andersfugmann","followers_url":"https://api.github.com/users/andersfugmann/followers","following_url":"https://api.github.com/users/andersfugmann/following{/other_user}","gists_url":"https://api.github.com/users/andersfugmann/gists{/gist_id}","starred_url":"https://api.github.com/users/andersfugmann/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/andersfugmann/subscriptions","organizations_url":"https://api.github.com/users/andersfugmann/orgs","repos_url":"https://api.github.com/users/andersfugmann/repos","events_url":"https://api.github.com/users/andersfugmann/events{/privacy}","received_events_url":"https://api.github.com/users/andersfugmann/received_events","type":"User","site_admin":false},"created_at":"2018-11-05T21:01:37Z","updated_at":"2018-11-05T21:01:37Z","author_association":"OWNER","body":"I've created a branch 'connection_on_closed', which should fix the problem.\r\nWould you mind testing that one, and report you observations?\r\n"}}
{"actor":{"display_login":"andersfugmann"}}