feat: fix after review

Author: sav-da

rabbitmq/functional_tests/basic_chaos/tests/test_rabbitmq.py

@@ -152,20 +152,27 @@ def message_consumed(data):
     assert response.status_code == 200
     await message_consumed.wait_call()
 
-    await gate.to_server_noop()
-    await gate.to_client_noop()
-    await asyncio.sleep(2.5)
-    await gate.to_server_pass()
-    await gate.to_client_pass()
-    await gate.sockets_close()
+    async with service_client.capture_logs(log_level='INFO') as capture:
 
-    await gate.wait_for_connections(timeout=10.0)
-    await asyncio.sleep(1.0)
+        @capture.subscribe(text="Consumer for queue 'chaos-queue' is broken, trying to restart")
+        def consumer_broken(**kwargs):
+            pass
 
-    response = await service_client.post('/v1/messages?message=after-heartbeat')
-    assert response.status_code == 200
+        @capture.subscribe(text='Restarted successfully')
+        def consumer_restarted(**kwargs):
+            pass
 
-    await message_consumed.wait_call()
+        await gate.to_server_drop()
+        await asyncio.sleep(3.0)
+        await gate.to_server_pass()
+
+        await consumer_broken.wait_call()
+        await consumer_restarted.wait_call()
+
+        response = await service_client.post('/v1/messages?message=after-heartbeat')
+        assert response.status_code == 200
+
+        await message_consumed.wait_call()
 
     response = await service_client.get('/v1/messages')
     assert response.status_code == 200

rabbitmq/src/urabbitmq/impl/amqp_connection_handler.cpp

@@ -180,6 +180,7 @@ void AmqpConnectionHandler::onReady(AMQP::Connection*) {
 }
 
 void AmqpConnectionHandler::OnConnectionCreated(AmqpConnection* connection, engine::Deadline deadline) {
+    UINVARIANT(connection_ == nullptr, "Unexpected repeated OnConnectionCreated call");
     connection_ = connection;
     reader_.Start(connection);
 

rabbitmq/src/urabbitmq/impl/header_value.cpp

@@ -85,7 +85,8 @@ AMQP::Array ToAmqpArray(const HeaderValue& value) {
 AMQP::Table ToAmqpTable(const HeaderValue& value) {
     AMQP::Table table;
     for (const auto& [key, item] : formats::common::Items(value)) {
-        WithAmqpField(item, [&table, &key](const AMQP::Field& field) { table.set(key, field); });
+        const auto key_copy = std::string{key};
+        WithAmqpField(item, [&table, key_copy](const AMQP::Field& field) { table.set(key_copy, field); });
     }
 
     return table;
@@ -157,7 +158,8 @@ std::unordered_map<std::string, HeaderValue> TableToHeaders(const AMQP::Table& t
 
 void AddHeadersToTable(AMQP::Table& table, const std::unordered_map<std::string, HeaderValue>& headers) {
     for (const auto& [key, value] : headers) {
-        WithAmqpField(value, [&table, &key](const AMQP::Field& field) { table.set(key, field); });
+        const auto key_copy = key;
+        WithAmqpField(value, [&table, key_copy](const AMQP::Field& field) { table.set(key_copy, field); });
     }
 }