sync

Author: goghcrow

.phpstorm.meta.php

@@ -3,7 +3,17 @@
 {
     $STATIC_METHOD_TYPES = [
         \Zan\Framework\Foundation\Container\Di::make('') => [
-            "" == "@",
-        ]
+            '' == '@',
+        ],
+
+//        new \ZanPHP\Contracts\Container\Container => [
+//            '' == '@',
+//        ],
+//        \ZanPHP\Contracts\Container\Container::make => [
+//            '' == '@',
+//        ],
+//        \make => [
+//            '' == '@',
+//        ]
     ];
 }

composer.json

@@ -9,23 +9,61 @@
             "type": "git",
             "url": "https://git.oschina.net/zan-group/nova-generic.git"
         },
+        "2": {
+            "type": "git",
+            "url": "https://git.oschina.net/zan-group/nova-generic.git"
+        },
+        "3": {
+            "type": "git",
+            "url": "https://github.com/zanphp/cache.git"
+        },
+	"4": {
+            "type": "git",
+            "url": "https://github.com/zanphp/container.git"
+        },
+	"5": {
+            "type": "git",
+            "url": "https://github.com/zanphp/contracts.git"
+        },
+	"6": {
+            "type": "git",
+            "url": "https://github.com/zanphp/etcd-client.git"
+        },
+	"7": {
+            "type": "git",
+            "url": "https://github.com/zanphp/loadbalance.git"
+        },
+	"8": {
+            "type": "git",
+            "url": "https://github.com/zanphp/service-chain.git"
+        },
         "packagist": {
             "type": "composer",
             "url": "https://packagist.phpcomposer.com"
         }
     },
     "require": {
-        "nova-service/generic": "dev-master",
-        "zanphp/nova": "dev-master",
         "packaged/thrift": "0.9.2.1",
         "symfony/console": "3.1",
-        "psr/log": "1.0.0"
+        "psr/log": "1.0.0",
+        "nova-service/generic": "dev-master",
+
+
+        "zanphp/cache": "dev-master",
+        "zanphp/container": "dev-master",
+        "zanphp/contracts": "dev-master",
+        "zanphp/etcd-client": "dev-master",
+        "zanphp/loadbalance": "dev-master",
+        "zanphp/nova": "dev-master",
+        "zanphp/service-chain": "dev-master"
+
     },
     "autoload": {
         "psr-4": {
             "Zan\\Framework\\": "src/"
         },
-        "classmap": [],
+        "classmap": [
+	],
         "files": [
             "src/Foundation/Coroutine/Command/Task.php",
             "src/Utilities/helpers.php"

src/Foundation/Application.php

@@ -6,6 +6,7 @@
 use Zan\Framework\Foundation\Booting\CheckIfBootable;
 use Zan\Framework\Foundation\Booting\InitializeCliInput;
 use Zan\Framework\Foundation\Booting\InitializeCache;
+use Zan\Framework\Foundation\Booting\InitializeContainer;
 use Zan\Framework\Foundation\Booting\InitializeKv;
 use Zan\Framework\Foundation\Booting\InitializeSyslog;
 use Zan\Framework\Foundation\Booting\LoadFiles;
@@ -97,6 +98,7 @@ protected function bootstrap()
             InitializePathes::class,
             LoadConfiguration::class,
             InitializeSharedObjects::class,
+            InitializeContainer::class,
             RegisterClassAliases::class,
             LoadFiles::class,
             InitializeCache::class,

src/Foundation/Booting/InitializeContainer.php

@@ -0,0 +1,26 @@
+<?php
+
+namespace Zan\Framework\Foundation\Booting;
+
+use Zan\Framework\Contract\Foundation\Bootable;
+use Zan\Framework\Foundation\Application;
+use Zan\Framework\Foundation\Core\Config;
+use ZanPHP\Container\Container;
+
+class InitializeContainer implements Bootable
+{
+
+    /**
+     * Bootstrap the given application.
+     *
+     * @param  \Zan\Framework\Foundation\Application $app
+     */
+    public function bootstrap(Application $app)
+    {
+        $binds = Config::get("zan_container", []);
+        $container = Container::getInstance();
+        foreach ($binds as $abstract => $bindArgs) {
+            $container->bind($abstract, ...(array)$bindArgs);
+        }
+    }
+}

src/Foundation/Coroutine/Command/Task.php

@@ -71,7 +71,7 @@ function getRpcContext($key = null, $default = null)
 {
     return new SysCall(function (Task $task) use($key, $default) {
         $context = $task->getContext();
-        $rpcCtx = $context->get(RpcContext::KEY, null, RpcContext::class);
+        $rpcCtx = $context->get("rpc-context", null, RpcContext::class);
         if ($rpcCtx) {
             $task->send($rpcCtx->get($key, $default));
         } else {
@@ -86,10 +86,10 @@ function setRpcContext($key, $value)
 {
     return new SysCall(function (Task $task) use ($key, $value) {
         $context = $task->getContext();
-        $rpcCtx = $context->get(RpcContext::KEY, null, RpcContext::class);
+        $rpcCtx = $context->get("rpc-context", null, RpcContext::class);
         if ($rpcCtx === null) {
             $rpcCtx = new RpcContext;
-            $context->set(RpcContext::KEY, $rpcCtx);
+            $context->set("rpc-context", $rpcCtx);
         }
         $task->send($rpcCtx->set($key, $value));
         return Signal::TASK_CONTINUE;
@@ -155,10 +155,10 @@ function wait()
     });
 }
 
-function parallel($coroutines)
+function parallel($coroutines, &$fetchCtx = [])
 {
-    return new SysCall(function (Task $task) use ($coroutines) {
-        (new Parallel($task))->call($coroutines);
+    return new SysCall(function (Task $task) use ($coroutines, &$fetchCtx) {
+        (new Parallel($task))->call($coroutines, $fetchCtx);
 
         return Signal::TASK_WAIT;
     });

src/Foundation/Coroutine/Parallel.php

@@ -4,23 +4,32 @@
 
 use Zan\Framework\Foundation\Exception\ParallelException;
 use Zan\Framework\Foundation\Exception\System\InvalidArgumentException;
+use Zan\Framework\Utilities\DesignPattern\Context;
 
 class Parallel
 {
     private $task;
     private $childTasks = [];
     private $sendValues = [];
     private $exceptions = [];
+    private $fetchCtx   = [];
+
+    /**
+     * @var Context
+     */
+    private $taskContext;
 
     public function __construct(Task $task)
     {
         $this->task = $task;
     }
 
-    public function call($coroutines)
+    public function call($coroutines, &$fetchCtx = [])
     {
+        $this->fetchCtx = &$fetchCtx;
+
         $parentTaskId = $this->task->getTaskId();
-        $taskContext = $this->task->getContext();
+        $this->taskContext = $this->task->getContext();
 
         $taskDoneEventName = 'parallel_task_done_' . $parentTaskId;
         $event = $this->task->getContext()->getEvent();
@@ -38,12 +47,14 @@ public function call($coroutines)
                 continue; 
             }
 
-            $childTask = new Task($this->catchException($key, $coroutine), $taskContext, 0, $this->task);
+            $childTask = new Task($this->catchException($key, $coroutine), $this->taskContext, 0, $this->task);
             $this->childTasks[$key] = $childTask;
 
             $newTaskId = $childTask->getTaskId();
             $evtName = 'task_event_' . $newTaskId;
             $eventChain->before($evtName, $taskDoneEventName);
+
+
         }
 
         if ($this->childTasks == []) {
@@ -85,14 +96,20 @@ public function done()
     private function catchException($key, \Generator $coroutine)
     {
         try {
-            yield $coroutine;
-            return;
-        } catch (\Throwable $t) {
-            $ex = t2ex($t);
-        } catch (\Exception $ex) { }
-
-        echo_exception($ex);
-        $this->exceptions[$key] = $ex;
-        yield $ex;
+            $r = (yield $coroutine);
+        } catch (\Throwable $r) {
+            $r = t2ex($r);
+            $this->exceptions[$key] = $r;
+            echo_exception($r);
+        } catch (\Exception $r) {
+            $this->exceptions[$key] = $r;
+            echo_exception($r);
+        }
+
+        foreach ($this->fetchCtx as $ctxKey) {
+            $this->fetchCtx[$ctxKey] = $this->taskContext->get($ctxKey);
+        }
+
+        yield $r;
     }
 }

src/Network/Common/HttpClient.php

@@ -10,6 +10,7 @@
 use Zan\Framework\Network\Common\Exception\HostNotFoundException;
 use Zan\Framework\Network\Server\Timer\Timer;
 use Zan\Framework\Network\Common\Exception\HttpClientTimeoutException;
+use Zan\Framework\Network\Tcp\RpcContext;
 use Zan\Framework\Sdk\Trace\Constant;
 use Zan\Framework\Sdk\Trace\DebuggerTrace;
 use Zan\Framework\Sdk\Trace\Trace;
@@ -41,6 +42,9 @@ class HttpClient implements Async
 
     private $callback;
 
+    /** @var RpcContext */
+    private $rpcContext;
+
     /** @var Trace */
     private $trace;
     private $traceHandle;
@@ -182,6 +186,7 @@ public function handle(Context $ctx = null)
         if ($ctx) {
             $this->trace = $ctx->get("trace");
             $this->debuggerTrace = $ctx->get('debugger_trace');
+            $this->rpcContext = $ctx->get("rpc-context");
         }
 
         if ($this->useHttpProxy) {
@@ -217,10 +222,8 @@ public function request($ip, $port)
         } else {
             $this->client = new \swoole_http_client($ip, $port, $this->ssl);
         }
+
         $this->buildHeader();
-        if (null !== $this->timeout) {
-            Timer::after($this->timeout, [$this, 'checkTimeout'], spl_object_hash($this));
-        }
 
         if ($this->trace) {
             $this->traceHandle = $this->trace->transactionBegin(Constant::HTTP_CALL, $this->host . $this->uri);
@@ -236,6 +239,10 @@ public function request($ip, $port)
             ]);
         }
 
+        if (null !== $this->timeout) {
+            Timer::after($this->timeout, [$this, 'checkTimeout'], spl_object_hash($this));
+        }
+
         if('GET' === $this->method){
             if ($this->trace) {
                 $this->trace->logEvent(Constant::GET, Constant::SUCCESS);
@@ -275,6 +282,13 @@ private function buildHeader()
             $this->header[DebuggerTrace::KEY] = $this->debuggerTrace->getKey();
         }
 
+        if ($this->rpcContext instanceof RpcContext) {
+            $pairs = $this->rpcContext->get();
+            foreach ($pairs as $key => $value) {
+                $this->header[$key] = $value;
+            }
+        }
+
         $this->client->setHeaders($this->header);
     }
 

src/Network/Connection/Driver/NovaClient.php

@@ -10,8 +10,9 @@
 use Kdt\Iron\Nova\Network\Client as NovaPingClient;
 use Kdt\Iron\Nova\Exception\NetworkException;
 use Zan\Framework\Utilities\Types\Time;
+use ZanPHP\Contracts\LoadBalance\Node;
 
-class NovaClient extends Base implements Connection
+class NovaClient extends Base implements Connection, Node
 {
     private $clientCb;
     protected $isAsync = true;
@@ -34,15 +35,17 @@ protected function closeSocket()
         }
     }
 
-    public function init() {
+    public function init()
+    {
         //set callback
         $this->getSocket()->on('connect', [$this, 'onConnect']);
         $this->getSocket()->on('receive', [$this, 'onReceive']);
         $this->getSocket()->on('close', [$this, 'onClose']);
         $this->getSocket()->on('error', [$this, 'onError']);
     }
 
-    public function onConnect(SwooleClient $cli) {
+    public function onConnect(SwooleClient $cli)
+    {
         //put conn to active_pool
         Timer::clearAfterJob($this->getConnectTimeoutJobId());
         Timer::clearAfterJob($this->getHeartbeatingJobId());
@@ -54,14 +57,16 @@ public function onConnect(SwooleClient $cli) {
         $this->inspect("connect to server", $cli);
     }
 
-    public function onClose(SwooleClient $cli){
+    public function onClose(SwooleClient $cli)
+    {
         Timer::clearAfterJob($this->getConnectTimeoutJobId());
         Timer::clearAfterJob($this->getHeartbeatingJobId());
         $this->close();
         $this->inspect("close", $cli);
     }
 
-    public function onReceive(SwooleClient $cli, $data) {
+    public function onReceive(SwooleClient $cli, $data)
+    {
         try {
             call_user_func($this->clientCb, $data);
         } catch (\Throwable $t) {
@@ -71,17 +76,20 @@ public function onReceive(SwooleClient $cli, $data) {
         }
     }
 
-    public function onError(SwooleClient $cli){
+    public function onError(SwooleClient $cli)
+    {
         Timer::clearAfterJob($this->getConnectTimeoutJobId());
         Timer::clearAfterJob($this->getHeartbeatingJobId());
         $this->close();
 
         $this->inspect("error", $cli, true);
     }
 
-    public function setClientCb(callable $cb) {
+    public function setClientCb(callable $cb)
+    {
         $this->clientCb = $cb;
     }
+
     public function heartbeat()
     {
         Timer::after($this->config['heartbeat-time'], [$this, 'heartbeating'], $this->getHeartbeatingJobId());
@@ -153,4 +161,13 @@ private function inspect($desc, SwooleClient $cli, $error = false)
 
         sys_echo("nova client $desc [" . implode(", ", $buffer) . "]");
     }
+
+    /**
+     * 0 ~ 100
+     * @return int|null
+     */
+    public function getWeight()
+    {
+        return isset($this->config["weight"]) ? $this->config["weight"] : 100;
+    }
 }