ex_nsq v1.1.0 NSQ.Consumer.Connections
Functions for connecting, disconnecting, managing connections, etc.
Link to this section Summary
Functions
When a connection is terminated or dies, we must do some extra cleanup.
First, a terminated process isn’t necessarily removed from the supervisor’s
list; therefore we call Supervisor.delete_child/2. And info about this
connection like RDY must be removed so it doesn’t contribute to total_rdy.
Not for external use
Create a connection to NSQD and add it to the consumer’s supervised list. Not for external use
Given a list of NSQD hosts, open a connection for each
We may have open connections which nsqlookupd stops reporting. This function tells us which connections we have stored in state but not in nsqlookupd. Not for external use
Iterate over all listed connections and delete the ones that are dead. This exists because it is difficult to reliably clean up a connection immediately after it is terminated (it might still be running). This function runs in the discovery loop to provide consistency
Frequently, when testing, we publish a message then immediately want a consumer to process it, but this doesn’t work if the consumer doesn’t discover the nsqd first. Only meant for testing
Finds and updates list of live NSQDs using either NSQ.Config.nsqlookupds or NSQ.Config.nsqds, depending on what’s configured. Preference is given to nsqlookupd. Not for external use
Initialized from NSQ.Consumer.Supervisor, sends the consumer a message on a fixed interval
Returns all live connections for a consumer. This function, which takes a consumer’s entire state as an argument, is for convenience. Not for external use
Returns all live connections for a consumer. Used in tests. Not for external use
When nsqlookupd reports available producers, there are some that may not already be in our connection list. This function reports which ones are new so we can connect to them
Given a single connection, immediately terminate its process (and all descendant processes, such as message handlers) and remove its info from the ConnInfo agent. Not for external use
Given a list of connections, force them to stop. Return the new state without those connections
Any inactive connections will be killed and any newly discovered connections will be added. Existing connections with no change are left alone. Not for external use
Link to this section Functions
cleanup_connection(pid, NSQ.Consumer.host_with_port, NSQ.Consumer.state) :: {:ok, NSQ.Consumer.state}
When a connection is terminated or dies, we must do some extra cleanup.
First, a terminated process isn’t necessarily removed from the supervisor’s
list; therefore we call Supervisor.delete_child/2. And info about this
connection like RDY must be removed so it doesn’t contribute to total_rdy.
Not for external use.
connect_to_nsqd(NSQ.Consumer.host_with_port, pid, NSQ.Consumer.state) :: {:ok, NSQ.Consumer.state}
Create a connection to NSQD and add it to the consumer’s supervised list. Not for external use.
connect_to_nsqds([NSQ.Consumer.host_with_port], pid, NSQ.Consumer.state) :: {:ok, NSQ.Consumer.state}
Given a list of NSQD hosts, open a connection for each.
dead_connections([NSQ.Consumer.host_with_port], pid, NSQ.Consumer.state) :: [NSQ.Consumer.connection]
We may have open connections which nsqlookupd stops reporting. This function tells us which connections we have stored in state but not in nsqlookupd. Not for external use.
delete_dead(NSQ.Consumer.state) :: {:ok, NSQ.Consumer.state}
Iterate over all listed connections and delete the ones that are dead. This exists because it is difficult to reliably clean up a connection immediately after it is terminated (it might still be running). This function runs in the discovery loop to provide consistency.
Frequently, when testing, we publish a message then immediately want a consumer to process it, but this doesn’t work if the consumer doesn’t discover the nsqd first. Only meant for testing.
discover_nsqds_and_connect(pid, NSQ.Consumer.state) :: {:ok, NSQ.Consumer.state}
Finds and updates list of live NSQDs using either NSQ.Config.nsqlookupds or NSQ.Config.nsqds, depending on what’s configured. Preference is given to nsqlookupd. Not for external use.
Initialized from NSQ.Consumer.Supervisor, sends the consumer a message on a fixed interval.
Returns all live connections for a consumer. This function, which takes a consumer’s entire state as an argument, is for convenience. Not for external use.
get(pid, NSQ.Consumer.state) :: [NSQ.Consumer.connection]
Returns all live connections for a consumer. Used in tests. Not for external use.
idle_with_rdy(NSQ.Consumer.state) :: [NSQ.Consumer.connection]
new_nsqds([NSQ.Consumer.host_with_port], pid, NSQ.Consumer.state) :: [NSQ.Consumer.host_with_port]
When nsqlookupd reports available producers, there are some that may not already be in our connection list. This function reports which ones are new so we can connect to them.
stop_connection(pid, NSQ.Consumer.host_with_port, NSQ.Consumer.state) :: {:ok, NSQ.Consumer.state}
Given a single connection, immediately terminate its process (and all descendant processes, such as message handlers) and remove its info from the ConnInfo agent. Not for external use.
stop_connections([NSQ.Consumer.connection], pid, NSQ.Consumer.state) :: {:ok, NSQ.Consumer.state}
Given a list of connections, force them to stop. Return the new state without those connections.
update([NSQ.Consumer.host_with_port], pid, NSQ.Consumer.state) :: {:ok, NSQ.Consumer.state}
Any inactive connections will be killed and any newly discovered connections will be added. Existing connections with no change are left alone. Not for external use.