Function get_from_buffer has a Cognitive Complexity of 47 (exceeds 5 allowed). Consider refactoring.

Open

def get_from_buffer(fields, prefix='', indent_level=2, remark=False):
    """
    Emit code that collects values from buf:offset, updating offset as progressing.
    :param remark: BE FUCKING VERBOSE! #DEBUG
    """

• Read up
  Read up

File consumer.py has 463 lines of code (exceeds 250 allowed). Consider refactoring.

Open

# coding=UTF-8
from __future__ import absolute_import, division, print_function

import io
import logging

Function get_serializer has a Cognitive Complexity of 46 (exceeds 5 allowed). Consider refactoring.

Open

def get_serializer(fields, prefix='', indent_level=2):  # type: (list, str) -> str, dict
    """
    Emit code that serializes the fields into buf at offset

    :param fields: list of Field instances

• Read up
  Read up

Function __acknack has a Cognitive Complexity of 39 (exceeds 5 allowed). Consider refactoring.

Open

    def __acknack(self, tag, multiple, ack):
        """
        :param tag: Note that 0 means "everything"
        :param ack: True to ack, False to nack
        """

• Read up
  Read up

Function on_body has a Cognitive Complexity of 29 (exceeds 5 allowed). Consider refactoring.

Open

    def on_body(self, payload):
        """:type payload: buffer"""
        assert self.state == 2
        self.data_to_go -= len(payload)

• Read up
  Read up

Function on_setup has a Cognitive Complexity of 29 (exceeds 5 allowed). Consider refactoring.

Open

    def on_setup(self, payload):  # type: (coolamqp.framing.base.AMQPMethodPayload) -> None
        """Called with different kinds of frames - during setup"""

        if isinstance(payload, ChannelOpenOk):
            # Do we need to declare the exchange?

• Read up
  Read up

Function _statemachine has a Cognitive Complexity of 28 (exceeds 5 allowed). Consider refactoring.

Open

    def _statemachine(self):
        # state rule 1
        if self.frame_type is None and self.total_data_len > 0:
            self.frame_type = self._extract_single_byte()

• Read up
  Read up

Function on_close has a Cognitive Complexity of 27 (exceeds 5 allowed). Consider refactoring.

Open

    def on_close(self, payload=None):
        # type: (tp.Optional[coolamqp.framing.base.AMQPMethodPayload]) -> None
        """
        Handle closing the channel. It sounds like an exception...

• Read up
  Read up

Function _pub has a Cognitive Complexity of 24 (exceeds 5 allowed). Consider refactoring.

Open

    def _pub(self, message, exchange_name, routing_key, parent_span=None, span_enqueued=None,
             dont_close_span=False):
        """
        Just send the message. Sends BasicDeliver + header + body.

• Read up
  Read up

File cluster.py has 293 lines of code (exceeds 250 allowed). Consider refactoring.

Open

# coding=UTF-8
from __future__ import print_function, absolute_import, division

import logging
import time

File connection.py has 285 lines of code (exceeds 250 allowed). Consider refactoring.

Open

# coding=UTF-8
from __future__ import absolute_import, division, print_function

import collections
import logging

File objects.py has 278 lines of code (exceeds 250 allowed). Consider refactoring.

Open

# coding=UTF-8
"""
Core objects used in CoolAMQP
"""
import logging

Function wait has a Cognitive Complexity of 19 (exceeds 5 allowed). Consider refactoring.

Open

    def wait(self, timeout=0.5):
        rds_and_exs = []        # waiting both for read and for exception
        wrs = []                # waiting for write
        for sock in six.itervalues(self.fd_to_sock):
            rds_and_exs.append(sock)

• Read up
  Read up

Function on_write has a Cognitive Complexity of 18 (exceeds 5 allowed). Consider refactoring.

Open

    def on_write(self):      # type: () -> None
        """
        Socket is writable, called by Listener

        :raises SocketFailed: on socket error

• Read up
  Read up

Function wait has a Cognitive Complexity of 17 (exceeds 5 allowed). Consider refactoring.

Open

    def wait(self, timeout=1):
        with self.socket_activation_lock:
            for socket_to_activate in self.sockets_to_activate:
                logger.debug('Activating fd %s', (socket_to_activate.fileno(),))
                self.epoll.register(socket_to_activate.fileno(), RW)

• Read up
  Read up

File publisher.py has 258 lines of code (exceeds 250 allowed). Consider refactoring.

Open

# coding=utf-8
"""
Module used to publish messages.

Expect wild NameErrors if you build this without RabbitMQ extensions (enabled by default),

Function _compile_particular_content_property_list_class has a Cognitive Complexity of 16 (exceeds 5 allowed). Consider refactoring.

Open

def _compile_particular_content_property_list_class(zpf, fields):
    """
    Compile a particular content property list.

    Particularity stems from

• Read up
  Read up

File declarer.py has 251 lines of code (exceeds 250 allowed). Consider refactoring.

Open

# coding=UTF-8
"""
queue.declare, exchange.declare and that shit
"""
from __future__ import print_function, absolute_import, division

Function publish has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

    def publish(self, message,  # type: Message
                exchange=None,  # type: tp.Union[Exchange, str, bytes]
                routing_key=u'',  # type: tp.Union[str, bytes]
                tx=None,  # type: tp.Optional[bool]
                confirm=None,  # type: tp.Optional[bool]

• Read up
  Read up

Function publish has 14 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def publish(self, message,  # type: Message