Skip to content

Commit 1005ad0

Browse files
authored
Basic_consume callback parameter should be called on_message_callback (#341)
* fixing issue with kwargs, when customers were calling basic_consume with kwargs the instrumentation was failing as the callback parameter was having a different name in the instrumentation * remove duplicated code * added popping queue, callback from args
1 parent f3d3ab8 commit 1005ad0

File tree

3 files changed

+11
-9
lines changed

3 files changed

+11
-9
lines changed

instana/instrumentation/pika.py

+9-7
Original file line numberDiff line numberDiff line change
@@ -76,8 +76,11 @@ def _bind_args(exchange, routing_key, body, properties=None, *args, **kwargs):
7676

7777

7878
def basic_get_with_instana(wrapped, instance, args, kwargs):
79-
def _bind_args(queue, callback, *args, **kwargs):
80-
return (queue, callback, args, kwargs)
79+
def _bind_args(*args, **kwargs):
80+
args = list(args)
81+
queue = kwargs.pop('queue', None) or args.pop(0)
82+
callback = kwargs.pop('callback', None) or kwargs.pop('on_message_callback', None) or args.pop(0)
83+
return (queue, callback, tuple(args), kwargs)
8184

8285
queue, callback, args, kwargs = _bind_args(*args, **kwargs)
8386

@@ -102,13 +105,12 @@ def _cb_wrapper(channel, method, properties, body):
102105
args = (queue, _cb_wrapper) + args
103106
return wrapped(*args, **kwargs)
104107

105-
106108
@wrapt.patch_function_wrapper('pika.adapters.blocking_connection', 'BlockingChannel.basic_consume')
107109
def basic_consume_with_instana(wrapped, instance, args, kwargs):
108-
def _bind_args(queue, on_consume_callback, *args, **kwargs):
109-
return (queue, on_consume_callback, args, kwargs)
110+
def _bind_args(queue, on_message_callback, *args, **kwargs):
111+
return (queue, on_message_callback, args, kwargs)
110112

111-
queue, on_consume_callback, args, kwargs = _bind_args(*args, **kwargs)
113+
queue, on_message_callback, args, kwargs = _bind_args(*args, **kwargs)
112114

113115
def _cb_wrapper(channel, method, properties, body):
114116
parent_span = tracer.extract(opentracing.Format.HTTP_HEADERS, properties.headers,
@@ -123,7 +125,7 @@ def _cb_wrapper(channel, method, properties, body):
123125
logger.debug("basic_consume_with_instana: ", exc_info=True)
124126

125127
try:
126-
on_consume_callback(channel, method, properties, body)
128+
on_message_callback(channel, method, properties, body)
127129
except Exception as e:
128130
scope.span.log_exception(e)
129131
raise

instana/version.py

+1-1
Original file line numberDiff line numberDiff line change
@@ -3,4 +3,4 @@
33

44
# Module version file. Used by setup.py and snapshot reporting.
55

6-
VERSION = '1.36.0'
6+
VERSION = '1.36.1'

tests/clients/test_pika.py

+1-1
Original file line numberDiff line numberDiff line change
@@ -246,7 +246,7 @@ def test_basic_consume_with_trace_context(self, _unused):
246246

247247
cb = mock.Mock()
248248

249-
self.obj.basic_consume("test.queue", cb, consumer_tag="test")
249+
self.obj.basic_consume(queue="test.queue", on_message_callback=cb, consumer_tag="test")
250250
self.obj._on_deliver(method_frame, header_frame, body)
251251

252252
spans = self.recorder.queued_spans()

0 commit comments

Comments
 (0)