public function run($force = false)
 {
     if ($force) {
         $this->killProgramsOnDreddPort();
     }
     $socket = sprintf('tcp://%s:%s', $this->host, $this->port);
     $server = stream_socket_server($socket, $errno, $errorMessage);
     if ($server === false) {
         throw new UnexpectedValueException("Server could not bind to socket: {$errorMessage}");
     }
     $buffer = "";
     for (;;) {
         $client = stream_socket_accept($server);
         while ($socketData = stream_socket_recvfrom($client, self::RECV_LENGTH)) {
             $buffer .= $socketData;
             // determine if message terminating character is present.
             if (strpos($buffer, self::MESSAGE_END) === false) {
                 continue;
             }
             $messages = [];
             foreach (explode(self::MESSAGE_END, $buffer) as $data) {
                 $message = json_decode($data);
                 // if not valid json the partial message needs saved
                 if (!$message) {
                     $buffer = $message;
                     continue;
                 }
                 $messages[] = $message;
             }
             foreach ($messages as $message) {
                 $this->processMessage($message, $client);
             }
         }
     }
 }

function sw_server_handle_recvfrom($server_socket, $events, $server)
{
    $data = stream_socket_recvfrom($server_socket, $server->buffer_size, $server->flags, $peer);
    if ($data !== false && $data != '') {
        $server->protocol->onData($peer, $data);
    }
}

 /**
  * {@inheritdoc}
  */
 public function read($length)
 {
     if (!$this->isReadable()) {
         throw new RuntimeException("Stream is not readable");
     }
     return stream_socket_recvfrom($this->getContext(), $length);
 }

    public function query($text)
    {
        $head = <<<_HEADER_
<?xml version="1.0" ?>
<wordsegmentation version="0.1">
<option showcategory="1" />
<authentication username="{$this->user}" password="{$this->passwd}" />
<text>
_HEADER_;
        $footer = <<<_FOOT_
</text>
</wordsegmentation>
_FOOT_;
        $this->data_send = $text;
        $text = str_replace("&", " ", $text);
        $querystr = $head . $text . $footer;
        $tempxml = simplexml_load_string($querystr);
        $resp = array();
        if ($tempxml) {
            if (stream_socket_sendto($this->sock, $tempxml->asXML())) {
                do {
                    $ttt = stream_socket_recvfrom($this->sock, 65525);
                    $ttt = iconv('big5', 'utf-8', $ttt);
                    $resp[] = $ttt;
                } while (!simplexml_load_string(implode($resp)));
                return $this->data_recv = html_entity_decode(implode($resp));
            }
        } else {
            $this->data_recv = 0;
            return null;
        }
    }

 /**
  * {@inheritdoc}
  */
 public function enableCrypto(int $method, float $timeout = 0) : \Generator
 {
     $resource = $this->getResource();
     if ($method & 1 || 0 === $method) {
         yield from $this->await($timeout);
     } else {
         yield from $this->poll($timeout);
         $raw = stream_socket_recvfrom($resource, 11, STREAM_PEEK);
         if (11 > strlen($raw)) {
             throw new FailureException('Failed to read crypto handshake.');
         }
         $data = unpack('ctype/nversion/nlength/Nembed/nmax-version', $raw);
         if (0x16 !== $data['type']) {
             throw new FailureException('Invalid crypto handshake.');
         }
         $version = $this->selectCryptoVersion($data['max-version']);
         if ($method & $version) {
             // Check if version was available in $method.
             $method = $version;
         }
     }
     do {
         // Error reporting suppressed since stream_socket_enable_crypto() emits E_WARNING on failure.
         $result = @stream_socket_enable_crypto($resource, (bool) $method, $method);
     } while (0 === $result && !(yield from $this->poll($timeout)));
     if ($result) {
         $this->crypto = $method;
         return;
     }
     $message = 'Failed to enable crypto.';
     if ($error = error_get_last()) {
         $message .= sprintf(' Errno: %d; %s', $error['type'], $error['message']);
     }
     throw new FailureException($message);
 }

 public function send($socket)
 {
     $deferred = new Deferred();
     stream_set_blocking($socket, false);
     $data = $this->getRequest();
     \Amp\onWritable($socket, function ($writer, $socket) use($deferred, &$data) {
         if ($bytes = fwrite($socket, $data)) {
             if ($bytes < \strlen($data)) {
                 $data = substr($data, $bytes);
                 return;
             }
             $size = 8192;
             \Amp\onReadable($socket, function ($reader, $socket) use($deferred, &$size) {
                 /* make attention to not read too much data */
                 $data = stream_socket_recvfrom($socket, $size, STREAM_PEEK);
                 if (false === ($pos = strpos($data, "\r\n\r\n"))) {
                     if (\strlen($data) == $size) {
                         $size *= 2;
                         // unbounded??
                     }
                     return;
                 }
                 \Amp\cancel($reader);
                 $deferred->succeed($this->parseResponse(fread($socket, $pos + 4)));
             });
         } else {
             $deferred->succeed(null);
         }
         \Amp\cancel($writer);
     });
     return $deferred->promise();
 }

 /**
  * @medium
  * @group connected
  * @expectedException \Predis\Protocol\ProtocolException
  */
 public function testThrowsExceptionOnProtocolDesynchronizationErrors()
 {
     $connection = $this->createConnection();
     $stream = $connection->getResource();
     $connection->writeRequest($this->getCurrentProfile()->createCommand('ping'));
     stream_socket_recvfrom($stream, 1);
     $connection->read();
 }

 public function handleData($stream)
 {
     $data = stream_socket_recvfrom($stream, $this->bufferSize);
     if ('' === $data || false === $data || feof($stream)) {
         $this->end();
     } else {
         $this->emit('data', array($data, $this));
     }
 }

 /** {@inheritdoc} */
 public function read(FramePickerInterface $picker, Context $context, $isOutOfBand)
 {
     stream_socket_recvfrom($this->socket->getStreamResource(), self::SOCKET_BUFFER_SIZE, STREAM_PEEK, $remoteAddress);
     if (!$remoteAddress && !$this->isLocalIo) {
         stream_socket_recvfrom($this->socket->getStreamResource(), self::SOCKET_BUFFER_SIZE);
         throw new AcceptException($this->socket, 'Can not accept client: failed to receive remote address.');
     }
     $reader = new DatagramClientIo($this->socket, $this->isLocalIo ? null : $remoteAddress);
     return new AcceptedFrame($remoteAddress, new UdpClientSocket($this->socket, $remoteAddress, $reader->read(new RawFramePicker(), $context, $isOutOfBand)));
 }

 public function handleData($socket)
 {
     $data = @stream_socket_recvfrom($socket, $this->bufferSize);
     if ('' === $data || false === $data) {
         $this->handleDisconnect($socket);
         $this->loop->removeStream($socket);
     } else {
         $client = $this->getClient($socket);
         $client->emit('data', array($data));
     }
 }

 function handleRecv($sock)
 {
     $pkt = stream_socket_recvfrom($sock, self::MAX_PACKET_SIZE, 0, $peer);
     if ($pkt == false) {
         $this->emit('error', array("Reading packet from {$peer} failed"));
         return;
     }
     if ($pkt != "") {
         $this->emit('packet', array($pkt, $peer, $this));
     }
 }

 function process()
 {
     $request = stream_socket_recvfrom($this->srv, 1500, 0, $this->cli);
     if ($request == false) {
         cy_log(CYE_ERROR, "read request from srv fd error.");
         return false;
     }
     $this->request_init();
     $this->request($request);
     $this->request_shutdown();
 }

 public function read($stream)
 {
     // Socket is raw, not using fread as it's interceptable by filters
     // See issues #192, #209, and #240
     $data = stream_socket_recvfrom($stream, $this->bufferSize);
     if ('' !== $data && false !== $data) {
         $this->notifyNext(new StreamEvent("/stream/data", $data));
     }
     if ('' === $data || false === $data || !is_resource($stream) || feof($stream)) {
         $this->notifyCompleted();
     }
 }

 public function server()
 {
     $socket = stream_socket_server("tcp://127.0.0.1:8000", $errno, $errstr, STREAM_SERVER_BIND);
     if (!$socket) {
         die("{$errstr} ({$errno})");
     }
     do {
         $pkt = stream_socket_recvfrom($socket, 1, 0, $peer);
         echo $pkt, "\n";
         echo "{$peer}\n";
         //stream_socket_sendto($socket, date("D M j H:i:s Y\r\n"), 0, $peer);
     } while (true);
 }

 public function read()
 {
     $data = stream_socket_recvfrom($this->socket, $this->bufferSize, 0, $peerAddress);
     if ($data === false) {
         // receiving data failed => remote side rejected one of our packets
         // due to the nature of UDP, there's no way to tell which one exactly
         // $peer is not filled either
         $this->notifyError(new \Exception('Invalid message'));
         return;
     }
     $this->notifyNext(new StreamEvent("/datagram/data", $data, ["peer" => $peerAddress]));
     $this->close();
 }

 /**
  * Accept a connection
  *
  * @return bool|mixed
  */
 public function accept()
 {
     if (($client = stream_socket_accept($this->socket)) === false) {
         // @codeCoverageIgnoreStart
         return false;
         // @codeCoverageIgnoreEnd
     }
     $data = '';
     while (($buf = stream_socket_recvfrom($client, 1024)) !== '') {
         $data .= $buf;
     }
     fclose($client);
     return unserialize($data);
 }

 function server_loop()
 {
     while (true) {
         $data = '';
         while (true) {
             $buf = stream_socket_recvfrom($this->server_sock, $this->buffer_size, 0, $peer);
             $data .= $buf;
             if ($buf === null or strlen($buf) < $this->buffer_size) {
                 break;
             }
         }
         $this->protocol->onData($peer, $data);
     }
 }

 /** {@inheritdoc} */
 protected function readRawDataIntoPicker(FramePickerInterface $picker, $isOutOfBand)
 {
     $size = self::SOCKET_BUFFER_SIZE;
     $resource = $this->socket->getStreamResource();
     do {
         $data = stream_socket_recvfrom($resource, $size, STREAM_PEEK);
         if (strlen($data) < $size) {
             break;
         }
         $size += $size;
     } while (true);
     $data = stream_socket_recvfrom($resource, $size, 0, $actualRemoteAddress);
     return $picker->pickUpData($data, $actualRemoteAddress);
 }

 /**
   Bi-directional UDP communication
   @param $msg the message
   @param $port integer port
   @return the response or an empty string
 
   Whatever program is listening on the other
   end cannot respond on the same port. It must
   send the response on (port+1).
 */
 public static function udpPoke($msg, $port = 9450)
 {
     $socket = stream_socket_server("udp://127.0.0.1:" . ($port + 1), $errno, $errstr, STREAM_SERVER_BIND);
     self::udpSend($msg, $port);
     $read = array($socket);
     $write = null;
     $except = null;
     $ready = stream_select($read, $write, $except, 0, 500);
     $buf = "";
     if ($ready > 0) {
         $buf = stream_socket_recvfrom($socket, 1024, 0, $peer);
     }
     stream_socket_shutdown($socket, STREAM_SHUT_RDWR);
     return $buf;
 }

 function run()
 {
     unlink(__DIR__ . '/alloc.log');
     unlink(__DIR__ . '/free.log');
     $socket = stream_socket_server("udp://127.0.0.1:9999", $errno, $errstr, STREAM_SERVER_BIND);
     if (!$socket) {
         die("{$errstr} ({$errno})");
     }
     while (1) {
         $pkt = stream_socket_recvfrom($socket, 65535, 0, $peer);
         $this->package_decode($pkt);
         //echo "$peer: $pkt\n";
         //stream_socket_sendto($socket, date("D M j H:i:s Y\r\n"), 0, $peer);
     }
 }