def __init__()

in python/rocketmq/publishing_message.py [0:0]

• 22 lines of code
• 12 McCabe index (conditional complexity)

    def __init__(self, message, publishing_settings, tx_enabled=False):
        self.message = message
        self.publishing_settings = publishing_settings
        self.tx_enabled = tx_enabled
        self.message_type = None

        max_body_size_bytes = publishing_settings.get_max_body_size_bytes()
        if len(message.body) > max_body_size_bytes:
            raise IOError(f"Message body size exceed the threshold, max size={max_body_size_bytes} bytes")

        self.message_id = MessageIdCodec.next_message_id()

        if not message.message_group and not message.delivery_timestamp and not tx_enabled:
            self.message_type = MessageType.NORMAL
            return

        if message.message_group and not tx_enabled:
            self.message_type = MessageType.FIFO
            return

        if message.delivery_timestamp and not tx_enabled:
            self.message_type = MessageType.DELAY
            return

        if message.message_group or message.delivery_timestamp or not tx_enabled:
            pass

        self.message_type = MessageType.TRANSACTION
        logger.debug(self.message_type)