Add shared AMQP connection registry with generation-based channel invalidation#117
Add shared AMQP connection registry with generation-based channel invalidation#117jwage wants to merge 7 commits into
Conversation
jwage
added
the
improvement
jwage
force-pushed
the
feature/connection-registry
branch
from
c26fd6b to
94be5f1
Compare
|
Would it be possible to get some help reviewing and testing this? @kraz @TamasSzigeti TamasSzigeti @dantleech |
kraz
left a comment
kraz
left a comment
There was a problem hiding this comment.
On the first sight, this new functionality is intended to be an "opt-in", but in reality the default state (none) actually changes how the connections are being managed, compared to before the connection_reuse was introduced.
Also, I remember there was a reported issue somewhere that when running the symfony messenger consumer from a frankenphp cli it forks the current process. So, it might be a good idea to use the process id as additional discriminator in AmqpConnectionRegistry::get along side with the generation so it does not reuse the same key.
| public function isConnected(): bool | ||
| { | ||
| return $this->connection !== null && $this->connection->isConnected(); | ||
| return $this->amqpConnectionRegistry | ||
| ->get($this->registryKey, $this->connectionConfig) | ||
| ->isConnected(); | ||
| } |
There was a problem hiding this comment.
This function should not introduce side effects. May be its better for the connection registry to have a has method, so it does not mutate its state unexpectedly.
| public function close(): void | ||
| { | ||
| $this->consumer?->stop(); | ||
| $this->connection?->close(); |
There was a problem hiding this comment.
Now all the connections will remain in memory until the process is terminated? This does not seem like an opt-in functionality.
| public function reconnect(): void | ||
| { | ||
| $this->connection?->reconnect(); | ||
| $this->channel = null; | ||
| $this->amqpConnectionRegistry->reconnect($this->registryKey, $this->connectionConfig); | ||
| $this->clearChannelState(); | ||
| $this->consumer?->stop(); | ||
| $this->consumer = null; | ||
| } |
There was a problem hiding this comment.
Even on the old variant of the code, I am wondering why the connection is reinitialized before stopping the consumer. In my opinion it might be better to first terminate the consumer and then reinitialize the connection:
try { $this->consumer?->stop(); } catch (\Throwable) {}
$this->consumer = null;
$this->amqpConnectionRegistry->reconnect($this->connectionIdentity, $this->connectionConfig);
$this->clearChannelState();2 participants
fixes #116