def message_from_packet()

in zktraffic/base/sniffer.py [0:0]

• 32 lines of code
• 6 McCabe index (conditional complexity)

  def message_from_packet(self, packet):
    """
    :returns: Returns an instance of ClientMessage or ServerMessage (or a subclass)
    :raises:
      :exc:`BadPacket` if the packet is for a client we are not tracking
      :exc:`DeserializationError` if deserialization failed
      :exc:`struct.error` if deserialization failed
    """
    client_port = self.config.client_port
    zk_port = self.config.zookeeper_port
    ip_p = get_ip_packet(packet.load, client_port, zk_port, self.config.is_loopback)

    if 0 == len(ip_p.data.data):
      return None

    if ip_p.data.dport == zk_port:
      data = ip_p.data.data
      src = intern("%s:%s" % (get_ip(ip_p, ip_p.src), ip_p.data.sport))
      dst = intern("%s:%s" % (get_ip(ip_p, ip_p.dst), ip_p.data.dport))
      client, server = src, dst
      if data.startswith(FOUR_LETTER_WORDS):
        self._set_four_letter_mode(client, data[0:4])
        raise BadPacket("Four letter request %s" % data[0:4])
      client_message = ClientMessage.from_payload(data, client, server)
      client_message.timestamp = packet.time
      self._track_client_message(client_message)
      return client_message

    if ip_p.data.sport == zk_port:
      data = ip_p.data.data
      src = intern("%s:%s" % (get_ip(ip_p, ip_p.src), ip_p.data.sport))
      dst = intern("%s:%s" % (get_ip(ip_p, ip_p.dst), ip_p.data.dport))
      client, server = dst, src
      four_letter = self._get_four_letter_mode(client)
      if four_letter:
        self._set_four_letter_mode(client, None)
        raise BadPacket("Four letter response %s" % four_letter)
      requests_xids = self._requests_xids.get(client, {})
      server_message = ServerMessage.from_payload(data, client, server, requests_xids)
      server_message.timestamp = packet.time
      return server_message

    raise BadPacket("Packet to the wrong port?")