def _regist_sniffer()

in zktraffic/omni/omni_sniffer.py [0:0]

• 16 lines of code
• 5 McCabe index (conditional complexity)

  def _regist_sniffer(self, ip, port, type):
    if (ip, port) in self._sniffers:
      current_sniffer = self._sniffers[(ip, port)]
      current_sniffer_type = self._get_sniffer_type(current_sniffer)
      assert type == current_sniffer_type, 'Conflict %s vs %s' % (type, current_sniffer_type)
      return

    if type == 'fle':
      sniffer = self.fle_sniffer_factory(port)
    elif type == 'zab':
      sniffer = self.zab_sniffer_factory(port)
    elif type == 'zk':
      sniffer = self.zk_sniffer_factory(port)
    else:
      raise ValueError('Unknown sniffer type %s' % type)
    print('OMNI DUMP REGISTERED SNIFFER %s(%s) (%s,%d)' % (sniffer, type, ip, port))
    self._sniffers[(ip, port)] = sniffer