On datagram receive.

        :param data:
        :param addr:
        :return:
425278
def datagram_received(self, data, addr):
        '''
        
        '''
        message = salt.utils.stringutils.to_unicode(data)
        if message.startswith(self.signature):
            try:
                timestamp = float(message[len(self.signature):])
            except (TypeError, ValueError):
                self.log.debug(
                    'Received invalid timestamp in package from %s:%s',
                    *addr
                )
                if self.disable_hidden:
                    self._sendto('{0}:E:{1}'.format(self.signature, 'Invalid timestamp'), addr)
                return

            if datetime.datetime.fromtimestamp(timestamp) < (datetime.datetime.now() - datetime.timedelta(seconds=20)):
                if self.disable_hidden:
                    self._sendto('{0}:E:{1}'.format(self.signature, 'Timestamp is too old'), addr)
                self.log.debug('Received outdated package from %s:%s', *addr)
                return

            self.log.debug('Received "%s" from %s:%s', message, *addr)
            self._sendto(
                salt.utils.stringutils.to_bytes(str('{0}:@:{1}').format(  # future lint: disable=blacklisted-function
                    self.signature,
                    salt.utils.json.dumps(self.answer, _json_module=_json)
                )),
                addr
            )
        else:
            if self.disable_hidden:
                self._sendto(
                    salt.utils.stringutils.to_bytes(
                        '{0}:E:{1}'.format(self.signature, 'Invalid packet signature'),
                        addr
                    )
                )
            self.log.debug('Received bad signature from %s:%s', *addr)
1066321
def send(self, data):
        """"""
        bytes_sent = self._sock.send(extract_bytes(data))
        self.bytes_written += bytes_sent
        return bytes_sent
1271891
def data(self, data):
        
        self._buffer = self._buffer + data

        while self._data_handler():
            pass
554703
def feed(self, data):
        
        self.buffer += data
        while len(self.buffer) >= 6:
            self.next_packet()
726817
def sendto(self, data, addr):
        
        asyncio.ensure_future(self.__inner_protocol.send_data(data, addr))
755413
def buffer(self, data):
        '''
        
        '''
        if not hasattr(self, '_sock'):
            return None

        # data will always be a byte array
        if len(self._buffer):
            self._buffer.extend(data)
        else:
            self._buffer = bytearray(data)
99004
def _send_receive(self, payload):
        
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        s.settimeout(self.connection_timeout)
        s.connect((self.address, self.port))
        s.send(payload)
        data = s.recv(1024)
        s.close()
        return data
490561
def receive_message(self, message, data):
        
        if data[MESSAGE_TYPE] == TYPE_STATUS:
            self._process_status(data.get("data"))
            return True

        return False
485341
def _send_packet(self, data):
        
        if self._closed:
            return

        data = json.dumps(data)

        def send():
            try:
                yield From(self.pipe_connection.write(data))
            except BrokenPipeError:
                self.detach_and_close()
        ensure_future(send())
1165069
def data_received(self, data):
        '''
        
        '''
        self._buffered_data.extend(data)
        while self._buffered_data:
            size = len(self._buffered_data)
            if self._data_package is None:
                if size < DataPackage.struct_datapackage.size:
                    return None
                self._data_package = DataPackage(self._buffered_data)
            if size < self._data_package.length:
                return None
            try:
                self._data_package.extract_data_from(self._buffered_data)
            except KeyError as e:
                logging.error('Unsupported package received: {}'.format(e))
            except Exception as e:
                logging.exception(e)
                # empty the byte-array to recover from this error
                self._buffered_data.clear()
            else:
                self._on_package_received()
            self._data_package = None
1213199
def on_packet_received(self, pkt):
        
        if not pkt:
            return
        if self.store:
            self.lst.append(pkt)
        if self.prn:
            result = self.prn(pkt)
            if result is not None:
                print(result)
783420
def send(self, data: bytes = b""):
        
        self.input.extend(data)
        self._process()
82546
def listen(self, timeout=10):
        
        self._socket.settimeout(float(timeout))
        while not self.stopped.isSet():
            try:
                data, client_address = self._socket.recvfrom(4096)
            except socket.timeout:
                continue
            try:

                self.receive_datagram((data, client_address))
            except RuntimeError:
                logger.exception("Exception with Executor")
        self._socket.close()
358448
def data(self, data):
        
        self.unpacker.feed(data)
        for msg in self.unpacker:
            self.handlers[msg[0]](*msg)
1176275
def send_datagram(self, message):
        
        if not self.stopped.isSet():
            host, port = message.destination
            logger.debug("send_datagram - " + str(message))
            serializer = Serializer()

            message = serializer.serialize(message)

            self._socket.sendto(message, (host, port))
480451
def receive_data(socket):
    
    answer = b""
    while True:
        packet = socket.recv(4096)
        if not packet: break
        answer += packet
    response = pickle.loads(answer)
    socket.close()
    return response
693605
def send(self, data):
        
        start = time.time()
        try:
            scgi_resp = ''.join(self.transport.send(_encode_payload(data)))
        finally:
            self.latency = time.time() - start

        resp, self.resp_headers = _parse_response(scgi_resp)
        return resp