def from_payload()

in zktraffic/fle/message.py [0:0]

• 35 lines of code
• 8 McCabe index (conditional complexity)

  def from_payload(cls, data, src, dst, timestamp):
    if len(data) < 16:
      raise BadPacket("Too small")

    proto, offset = read_long(data, 0)
    if proto == cls.PROTO_VER:
      server_id, offset = read_long(data, offset)
      election_addr, _ = read_string(data, offset)

      return Initial(timestamp, src, dst, server_id, election_addr)

    if len(data) >= cls.OLD_LEN:
      state, offset = read_number(data, 0)
      if PeerState.invalid(state):
        raise BadPacket("Invalid state: %d" % state)

      leader, offset = read_long(data, offset)
      zxid, offset = read_long(data, offset)
      election_epoch, offset = read_long(data, offset)
      peer_epoch, offset = read_long(data, offset) if len(data) > cls.OLD_LEN else (-1, offset)
      version = 0
      config = ""
      if len(data) > cls.WITH_VERSION_LEN:
        version, offset = read_number(data, offset)
        if version == 2:
          config, _ = read_string(data, offset)

      return Notification(
        timestamp,
        src,
        dst,
        state,
        leader,
        zxid,
        election_epoch,
        peer_epoch,
        version,
        config
      )

    raise BadPacket("Unknown unknown")