Atomic send of a string trough the socket

@param $msg

@return int
512807
protected function send($msg)
    {
        $msg .= PHP_EOL;

        return socket_write($this->socket, $msg, strlen($msg));
    }
640620
public function sendToAll(string $data, int $sender = 0, int $pageSize = 50): int
    {
        $startFd = 0;
        $count = 0;
        $fromUser = $sender < 1 ? 'SYSTEM' : $sender;
        $this->log("(broadcast)The #{$fromUser} send a message to all users. Data: {$data}");

        while (true) {
            $fdList = $this->server->connection_list($startFd, $pageSize);

            if ($fdList === false || ($num = \count($fdList)) === 0) {
                break;
            }

            $count += $num;
            $startFd = \end($fdList);

            /** @var $fdList array */
            foreach ($fdList as $fd) {
                $info = $this->getClientInfo($fd);

                if (isset($info['websocket_status']) && $info['websocket_status'] > 0) {
                    $this->server->push($fd, $data);
                }
            }
        }

        return $count;
    }
206272
public static function log($msg)
    {
        $msg = $msg . "\n";
        if (!self::$daemonize) {
            self::safeEcho($msg);
        }
        file_put_contents((string)self::$logFile, date('Y-m-d H:i:s') . ' ' . 'pid:' . getmypid() . ' ' . $msg, FILE_APPEND | LOCK_EX);
    }
44641
public function sendToWorkerHandler(MessageInterface $msg)
    {
        if (!$this->isWorkerHandlerReady()) {
            $this->getLogger()->debug('WorkerHandler socket not ready to send.');

            return;
        }

        $this->workerHandlerReady = false;
        $this->workerHandlerStream->send($msg);
    }
1036167
public function pack($msg)
    {
        if (!$msg instanceof Message) {
            trigger_error("Given parameter is not a message instance.",
                          E_USER_ERROR);
            return;
        }

        // Set value using serialzed message.
        $this->value = $msg->serializeToString();

        // Set type url.
        $pool = \Google\Protobuf\Internal\DescriptorPool::getGeneratedPool();
        $desc = $pool->getDescriptorByClassName(get_class($msg));
        $fully_qualifed_name = $desc->getFullName();
        $this->type_url = GPBUtil::TYPE_URL_PREFIX.substr(
            $fully_qualifed_name, 1, strlen($fully_qualifed_name));
    }
787765
function send($data) {
        $ps  = [];
        $ret = 0;
        foreach ($data as $p) {
            $ps[] = strlen($p);
            $ps[] = $p;
        }
        $s = join("\n", $ps) . "\n\n";
        if ($this->debug) {
            echo '> ' . str_replace(["\r", "\n"], ['\r', '\n'], $s) . "\n";
        }
        try {
            while (true) {
                $ret = @fwrite($this->sock, $s);
                if ($ret === false || $ret === 0) {
                    $this->close();
                    throw new SSDBException('Connection lost');
                }
                $s = substr($s, $ret);
                if (strlen($s) == 0) {
                    break;
                }
                @fflush($this->sock);
            }
        } catch (\Exception $e) {
            $this->close();
            throw new SSDBException($e->getMessage());
        }

        return $ret;
    }
782897
public function sendPong($msgid)
    {
        $messageNode = new ProtocolNode('iq',
            [
                'to'   => Constants::WHATSAPP_SERVER,
                'id'   => $msgid,
                'type' => 'result',
            ], null, '');

        $this->sendNode($messageNode);
        $this->eventManager()->fire('onSendPong',
            [
                $this->phoneNumber,
                $msgid,
            ]);
    }
327585
public function sendMsg($msg)
    {
        $apiUrl = 'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token='
            .$this->getAccessToken();
        $postStr = json_encode($msg, JSON_UNESCAPED_UNICODE);

        return $this->_curl($apiUrl, $postStr, 'post');
    }
552569
public function send(): int
    {
        $this->sent = true;
        $this->failedRecipients = [];
        return $this->mailer->send($this, $this->failedRecipients);
    }
928325
public function sendMessage($message)
    {
        $message = sprintf("%s\r\n", $message);
        fwrite($this->getSocketStream(), $message, strlen($message));
        return $this->receiveMessage();
    }
127977
static function message($msg)
    {
        $msg = empty($msg) ? '' : $msg;

        if(!is_string($msg)) {
            $msg = json_encode($msg);
        }

        echo $msg;

        return true;
    }
1298526
public function handlesMessage(Message $msg)
    {
        $handledMsgCodes = [
            Message::MSG_CALL,
            Message::MSG_CANCEL,
            Message::MSG_REGISTER,
            Message::MSG_UNREGISTER,
            Message::MSG_YIELD,
            Message::MSG_INTERRUPT
        ];

        if (in_array($msg->getMsgCode(), $handledMsgCodes)) {
            return true;
        } elseif ($msg instanceof ErrorMessage && $msg->getErrorMsgCode() === Message::MSG_INVOCATION) {
            return true;
        } elseif ($msg instanceof ErrorMessage && $msg->getErrorMsgCode() === Message::MSG_INTERRUPT) {
            return true;
        } else {
            return false;
        }
    }
752522
public function sendToUser($user_id, array $msg) {
        //检查redis ping 防止掉线
        $fid = $this->userIdToFid($user_id);
        if (!$fid) {
            return null;
        }
        $fid_server = explode('@', $fid);
        $host_name = $fid_server[ 0 ];
        $fid = $fid_server[ 1 ];
        if ($host_name == $this->host_name) {
            if (!$this->server->exist($fid)) {
                //断开
                $this->server->close($fid);
                $redis = Cache::redis();
                $redis->hDel('user_id_#_fid', $user_id);
                $redis->hDel('fid_#_user_id', $fid);
                Cache::release();
                return false;
            }
            $this->server->push($fid, json_encode($msg));
        } else {
            try {
                //通过集群中的其他服务器推送
                Http::put('http://' . $host_name . ':' . $this->config[ 'port' ] . '/open/push', [], ['fid' => $fid,
                                                                                                      'msg' => json_encode($msg),
                                                                                                      'secret' => $this->config[ 'secret' ]]);
            } catch (\Exception $exception) {
            } catch (\Error $exception) {
            }
        }
        return null;
    }
291434
public static function log($msg)
    {
        $msg = $msg."\n";
        if(static::$_status === static::STATUS_STARTING || !static::$daemonize)
        {
            echo $msg;
        }
        file_put_contents(static::$logFile, date('Y-m-d H:i:s') . " " . $msg, FILE_APPEND | LOCK_EX);
    }
1148456
public function sendMessage(Message $msg)
    {
        if (is_callable($this->sendMessageFunction)) {
            call_user_func_array($this->sendMessageFunction, [$msg]);
        }
    }
857562
protected function sendMessage($msg, $compress = true)
    {
        $this->sprintfIfDebug("sending '%s'\n", $msg);

        $msg = $compress ? $this->compress($msg) : $msg.PHP_EOL;
        $len = strlen($msg);

        $this->sprintfIfDebug("sending %d chars\n", $len);

        $blocks = ceil($len / self::BLOCK_SIZE);
        for ($i = 0; $i<$blocks; $i++) {
            $block = substr(
                $msg,
                $i * self::BLOCK_SIZE,
                ($i == $blocks-1) ? $len - ($i-1) * self::BLOCK_SIZE : self::BLOCK_SIZE
            );
            $prefix = sprintf("%06d:%03d:%03d:", strlen($block), $i+1, $blocks);
            $aux = $prefix.$block;

            $this->sprintfIfDebug("sending block %d of %d, prefix = %s\n", $i+1, $blocks, $prefix);

            if (false === $this->socket->write($aux, strlen($aux))) {
                throw new SocketException();
            };

            do {
                $read = $this->socket->read(10, PHP_NORMAL_READ);
            } while ($read && (strpos($read, self::ACK) !== 0));
        }

        return true;
    }
1151784
protected function send($data)
    {
        $this->debug("Send: $data");

        $result = $this->socket->write($data);
        if (!$result || PEAR::isError($result)) {
            $msg = $result ? $result->getMessage() : "unknown error";
            return PEAR::raiseError("Failed to write to socket: $msg");
        }

        return $result;
    }