Refactor events publisher

[moisses89]

Close #1814

General description

The focus of this pull request is to improve the communication between safe-transaction-service and the general-RabbitMQ, used as events broker.
The communication issues were:

• connection issues with RabbitMQ from celery task
• some events/message are lost when we have high volume of events to publish
• delayed events when we have high volume of events

Connection issues with RabbitMQ

After some research, we could see that the connection issues were shared socket resources between gevent greenlets from celery and pika greenlet, causing a loop of tries connecting and disconnecting.
ConcurrentObjectUseError('This socket is already used by another greenlet: <bound method Waiter.switch of <gevent._gevent_c_waiter.Waiter object at 0xffff9d30a9d0>>')

Some messages/events were lost

After create a test that send (delayed with celery) 10.000 of events with an incremental id and forcing to close the client connection from RabbitMQ management, I observed delayed event disconnection. During this delay several messages were sent without detecting the connection issue, more or less 500 events were lost.
This could be handled forcing ACK to each message from RabbitMQ, but I don't focus in this solution because the previous point.

Delayed events when we have high volume of events

Publish events from a task is slower than publish directly, the reason can be that to add a task that publish an event is as follows:

• publish to celery RabbitMQ the task to publish the event
• worker consume the task
• worker execute the task and publish to general-rabbitmq the event.
  vs:
• publish the event to general-rabbitMQ

Proposed solution on this PR

• use synchronous pika BlockingConnection instead asynchronous pika GeventConnection
• enable ACK to be able to resend any message that wasn't confirmed reception from RabbitMQ
• execute the publisher directly from signal instead execute it in a task
• because pika BlockingConnection cannot be shared between threads we must to create a pool of connections.

Benchmark

To do this benchmark we use asynchronous connection with ACK and synchronous connection with ACK.
The time calculation is the difference of instant time of last message sent and acknowledge less time instant of first message.
Asynchronous 100K messages takes 22 seconds
Synchronous 100K messages takes 27 seconds
It is important to mention and in asynchronous connection ACK is handled in other thread that means publish this quantity of messages takes around 800ms vs 27 seconds of BlockingConnection.

[Uxio0]

Suggested change

	if len(self.unsent_events):
	if self.unsent_events:

[Uxio0]

list(self.unsent_events)

This is to prevent race conditions?

[moisses89]

It was by mistake, I don't know why... but you touch a very interesting topic because two or more threads can call this function and duplicate events, I'll handle it.

[Uxio0]

Get the connection before unsent_events = self.unsent_events, in case there's a problem getting the connection

[moisses89]

Very good catch!

[moisses89]

updated https://github.com/safe-global/safe-transaction-service/pull/1818/files#diff-1c346d78e6f03c31bbf1285f635396fe0bdb24645164405ad615ad64e4524969R133