upload

.gitignore

@@ -0,0 +1,3 @@
+.idea
+composer.lock
+vendor

composer.json

@@ -0,0 +1,15 @@
+{
+  "name": "woon/workerman-rpc",
+  "require": {
+    "workerman/workerman": "^4.0",
+    "illuminate/support": "^8.60",
+    "ext-json": "*"
+  },
+  "autoload": {
+    "psr-4": {
+      "Rpc\\": "src",
+      "Protocols\\": "protocols",
+      "Test\\": "test"
+    }
+  }
+}

docker/.env

@@ -0,0 +1,12 @@
+### set local app dir there #######
+HOST_APP_DIR=../
+CONTAINER_APP_DIR=/usr/src/myapp
+HOST_PORT=8001
+CONTAINER_PORT=8001
+INSTALL_COMPOSER=true
+COMPOSER_INSTALLER=https://install.phpcomposer.com/installer
+COMPOSER_MIRROR=https://mirrors.aliyun.com/composer/
+INSTALL_LIB_EVENT=true
+CHANGE_AlPINE_SOURCE=true
+CONTAINER_PACKAGE_URL=mirrors.aliyun.com
+CONTAINER_NAME=workerman-rpc

docker/Dockerfile

@@ -0,0 +1,77 @@
+FROM php:7.4-cli-alpine3.13
+# Set working dir.
+ARG CONTAINER_APP_DIR
+
+ARG timezone
+
+ENV TIMEZONE=${timezone:-"Asia/Shanghai"} \
+    APP_ENV=prod \
+    SCAN_CACHEABLE=(true)
+
+
+# Change alpinelinux source.
+# Reference: https://github.com/laradock/laradock
+ARG CHANGE_AlPINE_SOURCE
+ARG CONTAINER_PACKAGE_URL
+RUN if [ $CHANGE_AlPINE_SOURCE ] ; then sed -i "s/dl-cdn.alpinelinux.org/${CONTAINER_PACKAGE_URL}/g" /etc/apk/repositories ; fi
+
+RUN uname -a && \
+    apk update
+
+RUN docker-php-ext-install sockets pcntl && \
+    docker-php-ext-install pdo_mysql
+
+RUN apk add --no-cache pcre-dev $PHPIZE_DEPS && \
+    pecl install redis && \
+    docker-php-ext-enable redis
+
+ARG INSTALL_LIB_EVENT
+RUN if [ ${INSTALL_LIB_EVENT} = true ]; then \
+    apk add libevent-dev openssl-dev libressl-dev && \
+    pecl install event && \
+    echo extension=event.so > /usr/local/etc/php/conf.d/event.ini \
+;fi
+
+# update
+RUN set -ex \
+    # show php version and extensions
+    && php -v \
+    && php -m \
+    #  ---------- some config ----------
+    && cd /usr/local/etc/php \
+    # - config PHP
+    && { \
+        echo "upload_max_filesize=128M"; \
+        echo "post_max_size=128M"; \
+        echo "memory_limit=1G"; \
+        echo "date.timezone=${TIMEZONE}"; \
+    } | tee conf.d/99_overrides.ini \
+    # - config timezone
+    && ln -sf /usr/share/zoneinfo/${TIMEZONE} /etc/localtime \
+    && echo "${TIMEZONE}" > /etc/timezone \
+    # ---------- clear works ----------
+    && rm -rf /var/cache/apk/* /tmp/* /usr/share/man \
+    && echo -e "\033[42;37m Build Completed :).\033[0m\n"
+
+# Install composer and set mirror.
+# Reference: https://pkg.phpcomposer.com/#how-to-install-composer
+# Referencr: https://developer.aliyun.com/composer
+ARG INSTALL_COMPOSER
+ARG COMPOSER_INSTALLER
+ARG COMPOSER_MIRROR
+RUN if [ ${INSTALL_COMPOSER} = true ]; then \
+    php -r "copy('${COMPOSER_INSTALLER}', 'composer-setup.php');" && \
+    php composer-setup.php && \
+    php -r "unlink('composer-setup.php');" && \
+    mv composer.phar /usr/local/bin/composer && \
+    composer config -g repo.packagist composer ${COMPOSER_MIRROR} \
+;fi
+
+
+RUN apk add git zip vim bash
+
+# Expose port
+ARG CONTAINER_PORT
+EXPOSE ${CONTAINER_PORT}
+
+WORKDIR ${CONTAINER_APP_DIR}

docker/Readme.md

@@ -0,0 +1,24 @@
+# Set environment variables
+Copy file **env-example** to **.env**, and change the item what you need.
+
+Change the **HOST_APP_DIR** env item, ensure where your webman app is. 
+
+# Build image
+Ensure you have install docker and docker-compose first.
+
+You can build you image by execute follow command:
+
+> **docker-compose up -d webman**
+
+# Run
+Enter the container:
+> **docker-compose exec webman bash**
+
+If you havn't init composer, you can run this command:
+> **composer install**
+
+Run in front: 
+> **php start.php start**
+
+Run in backend:
+> **php start.php start -d**

docker/docker-compose.yaml

@@ -0,0 +1,28 @@
+version: '3'
+services:
+  workerman-rpc:
+    build:
+      context: ./
+      args:
+        - CONTAINER_PORT=${CONTAINER_PORT}
+        - CONTAINER_APP_DIR=${CONTAINER_APP_DIR}
+        - INSTALL_COMPOSER=${INSTALL_COMPOSER}
+        - COMPOSER_INSTALLER=${COMPOSER_INSTALLER}
+        - COMPOSER_MIRROR=${COMPOSER_MIRROR}
+        - INSTALL_LIB_EVENT=${INSTALL_LIB_EVENT}
+        - CONTAINER_PACKAGE_URL=${CONTAINER_PACKAGE_URL}
+        - CHANGE_AlPINE_SOURCE=${CHANGE_AlPINE_SOURCE}
+    ports:
+      - "9501:9501"
+    volumes:
+      - ${HOST_APP_DIR}:${CONTAINER_APP_DIR}
+    container_name: ${CONTAINER_NAME}
+    networks:
+      - network
+    stdin_open: true
+    tty: true
+    extra_hosts:
+      - "host.docker.internal:host-gateway"
+
+networks:
+  network:

docker/env-example

@@ -0,0 +1,11 @@
+### set local app dir there #######
+HOST_APP_DIR=../webman
+CONTAINER_APP_DIR=/usr/src/myapp
+HOST_PORT=8787
+CONTAINER_PORT=8787
+INSTALL_COMPOSER=true
+COMPOSER_INSTALLER=https://install.phpcomposer.com/installer
+COMPOSER_MIRROR=https://mirrors.aliyun.com/composer/
+INSTALL_LIB_EVENT=true
+CHANGE_UBUNTU_SOURCE=true
+UBUNTU_SOURCE=tsinghua

docker/sources.sh

@@ -0,0 +1,40 @@
+#!/bin/bash
+# Reference: https://github.com/laradock/laradock
+set -xe;
+if type "tee" 2>/dev/null && [ -n "${UBUNTU_SOURCE}" ]; then
+    SOURCE_PATH="/etc/apt/sources.list"
+    cp ${SOURCE_PATH} ${SOURCE_PATH}.bak && rm -rf ${SOURCE_PATH}
+    case "${UBUNTU_SOURCE}" in
+        "aliyun")
+            tee ${SOURCE_PATH} <<-'EOF'
+deb http://mirrors.aliyun.com/ubuntu/ bionic main restricted universe multiverse
+deb http://mirrors.aliyun.com/ubuntu/ bionic-security main restricted universe multiverse
+deb http://mirrors.aliyun.com/ubuntu/ bionic-updates main restricted universe multiverse
+deb http://mirrors.aliyun.com/ubuntu/ bionic-proposed main restricted universe multiverse
+deb http://mirrors.aliyun.com/ubuntu/ bionic-backports main restricted universe multiverse
+deb-src http://mirrors.aliyun.com/ubuntu/ bionic main restricted universe multiverse
+deb-src http://mirrors.aliyun.com/ubuntu/ bionic-security main restricted universe multiverse
+deb-src http://mirrors.aliyun.com/ubuntu/ bionic-updates main restricted universe multiverse
+deb-src http://mirrors.aliyun.com/ubuntu/ bionic-proposed main restricted universe multiverse
+deb-src http://mirrors.aliyun.com/ubuntu/ bionic-backports main restricted universe multiverse
+EOF
+;;
+        "tsinghua")
+            tee ${SOURCE_PATH} <<-'EOF'
+deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu/ bionic main restricted universe multiverse
+deb-src https://mirrors.tuna.tsinghua.edu.cn/ubuntu/ bionic main restricted universe multiverse
+deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu/ bionic-updates main restricted universe multiverse
+deb-src https://mirrors.tuna.tsinghua.edu.cn/ubuntu/ bionic-updates main restricted universe multiverse
+deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu/ bionic-backports main restricted universe multiverse
+deb-src https://mirrors.tuna.tsinghua.edu.cn/ubuntu/ bionic-backports main restricted universe multiverse
+deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu/ bionic-security main restricted universe multiverse
+deb-src https://mirrors.tuna.tsinghua.edu.cn/ubuntu/ bionic-security main restricted universe multiverse
+deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu/ bionic-proposed main restricted universe multiverse
+deb-src https://mirrors.tuna.tsinghua.edu.cn/ubuntu/ bionic-proposed main restricted universe multiverse
+EOF
+;;
+        *)
+            echo "There is no ubuntu source detected!"
+            exit 1;;
+    esac
+fi

protocols/Json.php

@@ -0,0 +1,36 @@
+<?php
+/**
+ *
+ * @author jhx
+ * @date 2021/10/20 9:37
+ */
+
+namespace Protocols;
+
+
+class Json
+{
+    public static function input($buffer)
+    {
+        // 获得换行字符"\n"位置
+        $pos = strpos($buffer, "\n");
+        // 没有换行符，无法得知包长，返回0继续等待数据
+        if ($pos === false) {
+            return 0;
+        }
+        // 有换行符，返回当前包长（包含换行符）
+        return $pos + 1;
+    }
+
+    public static function encode($buffer)
+    {
+        // json序列化，并加上换行符作为请求结束的标记
+        return json_encode($buffer) . "\n";
+    }
+
+    public static function decode($buffer)
+    {
+        // 去掉换行，还原成数组
+        return json_decode(trim($buffer), true);
+    }
+}

src/RpcClient.php

@@ -0,0 +1,171 @@
+<?php
+/**
+ *
+ * @author jhx
+ * @date 2021/10/20 10:33
+ */
+
+namespace Rpc;
+
+
+use Illuminate\Support\Arr;
+use Protocols\Json;
+
+class RpcClient
+{
+    private $timeout = 5;
+
+    /**
+     * 异步调用发送数据前缀
+     * @var string
+     */
+    const ASYNC_SEND_PREFIX = 'asend_';
+
+    /**
+     * 异步调用接收数据
+     * @var string
+     */
+    const ASYNC_RECV_PREFIX = 'arecv_';
+
+    private $addressArray = [];
+
+    private $asyncInstances = [];
+
+    private static $instances = [];
+
+    private $connection;
+
+    private $server;
+
+    private $className;
+
+    private $user = '';
+
+    private $passwd = '';
+
+    public function __construct($server, $className)
+    {
+        $this->server    = $server;
+        $this->className = $className;
+    }
+
+    /**
+     * 防止对象实例被克隆
+     *
+     * @return void
+     */
+    private function __clone()
+    {
+    }
+
+    /**
+     * 防止被反序列化
+     *
+     * @return void
+     */
+    private function __wakeup()
+    {
+    }
+
+    public static function instance($server, $className)
+    {
+        if (!isset(self::$instances[$server])) {
+            self::$instances[$server][$className] = new self($server, $className);
+        }
+
+        return self::$instances[$server][$className];
+    }
+
+    public function address($addressArray)
+    {
+        $this->addressArray = $addressArray;
+        return $this;
+    }
+
+    public function __call($method, $arguments)
+    {
+        // 判断是否是异步发送
+        if (0 === strpos($method, self::ASYNC_SEND_PREFIX)) {
+            $realMethod  = substr($method, strlen(self::ASYNC_SEND_PREFIX));
+            $instanceKey = $realMethod . serialize($arguments);
+            if (isset($this->$asyncInstances[$instanceKey])) {
+                throw new \Exception($this->className . "->$method(" . implode(',', $arguments) . ") have already been called");
+            }
+            $this->asyncInstances[$instanceKey] = new self($this->server, $this->className);
+            return $this->asyncInstances[$instanceKey]->sendData($realMethod, $arguments);
+        }
+        // 如果是异步接受数据
+        if (0 === strpos($method, self::ASYNC_RECV_PREFIX)) {
+            $realMethod  = substr($method, strlen(self::ASYNC_RECV_PREFIX));
+            $instanceKey = $realMethod . serialize($arguments);
+            if (!isset($this->asyncInstances[$instanceKey])) {
+                throw new \Exception($this->server . "@" . $this->className . "->asend_$realMethod(" . implode(',', $arguments) . ") have not been called");
+            }
+            $tmp = $this->asyncInstances[$instanceKey];
+            unset($this->asyncInstances[$instanceKey]);
+            return $tmp->recvData();
+        }
+        // 同步发送接收
+        $this->sendData($method, $arguments);
+        return $this->recvData();
+    }
+
+    public function sendData($method, $arguments)
+    {
+        $this->openConnection();
+
+        $binData = Json::encode(array(
+            'class'       => $this->className,
+            'method'      => $method,
+            'param_array' => $arguments,
+            'user'        => $this->user,
+            'passwd'      => $this->passwd
+        ));
+
+        if (fwrite($this->connection, $binData) !== strlen($binData)) {
+            throw new \Exception('Can not send data');
+        }
+        return true;
+    }
+
+    public function recvData()
+    {
+        $ret = fgets($this->connection);
+        $this->closeConnection();
+        if (!$ret) {
+            throw new \Exception("recvData empty");
+        }
+        return Json::decode($ret);
+    }
+
+    protected function openConnection()
+    {
+        $addressArr = $this->addressArray;
+        if (empty($addressArr)) {
+            throw new \Exception("addressArr empty");
+        }
+
+        $address = $addressArr[array_rand($addressArr)];
+
+        $addressArr = explode('@', $address);
+
+        if (isset($addressArr[1])) {
+            $account      = explode(':', $addressArr[1]);
+            $this->user   = Arr::get($account, 0, '');
+            $this->passwd = Arr::get($account, 1, '');
+        }
+
+        $this->connection = stream_socket_client($address, $errNo, $errMsg);
+        if (!$this->connection) {
+            throw new \Exception("can not connect to $address , $errNo:$errMsg");
+        }
+        stream_set_blocking($this->connection, true);
+        stream_set_timeout($this->connection, $this->timeout);
+    }
+
+    protected function closeConnection()
+    {
+        fclose($this->connection);
+        $this->connection = null;
+    }
+}

src/RpcServer.php

@@ -0,0 +1,101 @@
+<?php
+/**
+ *
+ * @author jhx
+ * @date 2021/10/20 8:54
+ */
+
+namespace Rpc;
+
+
+use Illuminate\Support\Arr;
+use Workerman\Worker;
+
+class RpcServer
+{
+    private $config;
+
+    public function __construct()
+    {
+
+    }
+
+    /**
+     * @return mixed
+     */
+    public function getConfig()
+    {
+        return $this->config;
+    }
+
+    /**
+     * @param mixed $config
+     */
+    public function setConfig($config)
+    {
+        $this->config = $config;
+        return $this;
+    }
+
+    public function createWorker()
+    {
+        $socketName = Arr::get($this->getConfig(), 'socket_name', 'json://0.0.0.0:9501');
+        $worker     = new Worker($socketName);
+
+        $worker->count = Arr::get($this->getConfig(), 'process_count', 1);
+
+        $worker->name = Arr::get($this->getConfig(), 'worker_name', 'rpc_server');
+
+        $worker->onMessage = function ($connection, $data) {
+
+            if (empty($data['class']) || empty($data['method']) || !isset($data['param_array'])) {
+                // 发送数据给客户端，请求包错误
+                return $connection->send(['code' => 400, 'msg' => 'bad request', 'data' => null]);
+            }
+
+            $servicePath = Arr::get($this->getConfig(), 'service_path');
+            $class       = Arr::get($data, 'class');
+            $method      = Arr::get($data, 'method');
+            $paramArray  = Arr::get($data, 'param_array');
+
+            // 账号密码
+            $account = Arr::get($this->getConfig(), 'account', []);
+            if (!empty($account)) {
+                $user   = Arr::get($data, 'user');
+                $passwd = Arr::get($data, 'passwd');
+                if (empty($user) || empty($passwd) || !array_key_exists($user, $account) || ($passwd != Arr::get($account, $user))) {
+                    return $connection->send(['code' => 400, 'msg' => 'user or passwd is error', 'data' => null]);
+                }
+            }
+
+            $class = "{$servicePath}\\{$class}";
+
+            // 判断类对应文件是否载入
+            if (class_exists($class) && method_exists($class, $method)) {
+                try {
+                    $class = new $class;
+                    $res   = call_user_func_array([$class, $method], $paramArray);
+                    return $connection->send(['code' => 0, 'msg' => 'ok', 'data' => $res]);
+                } catch (\Exception $e) {
+                    $code = $e->getCode() ? $e->getCode() : 500;
+                    return $connection->send(['code' => $code, 'msg' => $e->getMessage(), 'data' => $e]);
+                }
+            }
+
+            $code = 404;
+            $msg  = "class $class or method $method not found";
+            // 发送数据给客户端 类不存在
+            return $connection->send(['code' => $code, 'msg' => $msg, 'data' => null]);
+
+        };
+
+        return $this;
+    }
+
+    public function run()
+    {
+        Worker::runAll();
+    }
+
+
+}

test/Client.php

@@ -0,0 +1,30 @@
+<?php
+define('VENDOR_PATH', __DIR__ . '/../vendor/autoload.php');
+
+require_once VENDOR_PATH;
+
+ini_set('date.timezone', 'Asia/Shanghai');
+
+$addressArray = [
+    'tcp://127.0.0.1:9501@user1:passwd1',
+    'tcp://127.0.0.1:9501@user1:passwd2'
+];
+
+$TestClient = \Rpc\RpcClient::instance('test', 'TestService');
+$TestClient = $TestClient->address($addressArray);
+$retSync    = $TestClient->test('test_1');
+
+
+$retAsync = $TestClient->asend_test('test_2');
+$retAsync = $TestClient->arecv_test('test_2');
+
+
+var_dump($retSync);
+var_dump($retAsync);
+
+
+$userClient = \Rpc\RpcClient::instance('test', 'UserService');
+$userClient = $userClient->address($addressArray);
+$retSync    = $userClient->test('test_1');
+
+var_dump($retSync);

test/Server.php

@@ -0,0 +1,18 @@
+<?php
+define('VENDOR_PATH', __DIR__ . '/../vendor/autoload.php');
+
+require_once VENDOR_PATH;
+
+ini_set('date.timezone', 'Asia/Shanghai');
+
+
+$config = [
+    'service_path'  => "Test\\Service",
+    'process_count' => 10,
+    'account'       => [
+        'user1' => 'passwd1'
+    ]
+];
+
+
+(new \Rpc\RpcServer())->setConfig($config)->createWorker()->run();

test/service/TestService.php

@@ -0,0 +1,17 @@
+<?php
+/**
+ *
+ * @author jhx
+ * @date 2021/10/20 11:25
+ */
+
+namespace Test\service;
+
+
+class TestService
+{
+    public function test($string)
+    {
+        return "hello {$string}";
+    }
+}