基于 Broker + NATS 在客户端和服务端实现基于事件驱动的异步通信（下）

app/Console/Commands/ProcessBrokerMessage.php

@@ -0,0 +1,57 @@
+<?php
+namespace App\Console\Commands;
+
+use App\MicroApi\Items\UserItem;
+use App\MicroApi\Services\UserService;
+use App\Services\Broker\BrokerService;
+use Illuminate\Console\Command;
+
+class ProcessBrokerMessage extends Command
+{
+    protected $signature = 'process:broker-message';
+
+    protected $description = 'Subscribe and process message from micro broker';
+
+    /**
+     * @var UserService
+     */
+    protected $userService;
+
+    /**
+     * Create a new command instance.
+     * @return void
+     */
+    public function __construct()
+    {
+        parent::__construct();
+        $this->userService = resolve("microUserService");
+    }
+
+    /**
+     * Execute the console command.
+     *
+     * @return mixed
+     */
+    public function handle()
+    {
+        $broker = new BrokerService();
+        $broker->subscribe('password.reset', function ($message) {
+            // 解析消息数据
+            $message = $message->getBody();
+            $passwordReset = json_decode(base64_decode($message['Body']));
+            $email = $passwordReset->email;
+            $token = $passwordReset->token;
+            // 发送重置邮件
+            $user = $this->userService->getByEmail($email);
+            if ($user) {
+                $model = new UserItem();
+                $model->fillAttributes($user);
+                $model->sendPasswordResetNotification($token);
+                $this->info('密码重置邮件已发送[email:' . $email . ']');
+            } else {
+                $this->error('指定用户不存在[email:' . $email . ']');
+            }
+        });
+        $broker->wait();
+    }
+}

app/Console/Kernel.php

@@ -4,6 +4,7 @@ namespace App\Console;
 
 use Illuminate\Console\Scheduling\Schedule;
 use Illuminate\Foundation\Console\Kernel as ConsoleKernel;
+use App\Console\Commands\ProcessBrokerMessage;
 
 class Kernel extends ConsoleKernel
 {
@@ -13,7 +14,8 @@ class Kernel extends ConsoleKernel
      * @var array
      */
     protected $commands = [
-        Commands\LaraStructure::class
+        Commands\LaraStructure::class,
+        ProcessBrokerMessage::class,
     ];
 
     /**

app/Services/Auth/PasswordBroker.php

@@ -0,0 +1,25 @@
+<?php
+namespace App\Services\Auth;
+
+use Illuminate\Auth\Passwords\PasswordBroker as BasePasswordBroker;
+
+class PasswordBroker extends BasePasswordBroker
+{
+    /**
+     * Send a password reset link to a user.
+     *
+     * @param  array  $credentials
+     * @return string
+     */
+    public function sendResetLink(array $credentials)
+    {
+        // 检查用户是否存在
+        $user = $this->getUser($credentials);
+        if (is_null($user)) {
+            return static::INVALID_USER;
+        }
+        // 存在的话则创建对应的密码重置记录，邮件发送操作异步去做
+        $this->tokens->create($user);
+        return static::RESET_LINK_SENT;
+    }
+}

app/Services/Auth/PasswordBrokerManager.php

@@ -2,6 +2,7 @@
 namespace App\Services\Auth;
 
 use Illuminate\Auth\Passwords\PasswordBrokerManager as BasePasswordBrokerManager;
+use InvalidArgumentException;
 
 class PasswordBrokerManager extends BasePasswordBrokerManager
 {
@@ -15,4 +16,25 @@ class PasswordBrokerManager extends BasePasswordBrokerManager
     {
         return new ServiceTokenRepository();
     }
+    /**
+     * Resolve the given broker.
+     *
+     * @param  string  $name
+     * @return PasswordBroker
+     *
+     * @throws InvalidArgumentException
+     */
+    protected function resolve($name)
+    {
+        $config = $this->getConfig($name);
+
+        if (is_null($config)) {
+            throw new InvalidArgumentException("密码重置器 [{$name}] 未定义");
+        }
+
+        return new PasswordBroker(
+            $this->createTokenRepository($config),
+            $this->app['auth']->createUserProvider($config['provider'] ?? null)
+        );
+    }
 }

app/Services/Broker/BrokerService.php

@@ -0,0 +1,49 @@
+<?php
+namespace App\Services\Broker;
+
+use Nats\ConnectionOptions;
+use Nats\EncodedConnection;
+use Nats\Encoders\JSONEncoder;
+
+class BrokerService
+{
+    /**
+     * @var EncodedConnection
+     */
+    protected $client;
+
+    public function __construct()
+    {
+        $encoder = new JSONEncoder();
+        $options = new ConnectionOptions([
+            'host' => config('services.micro.broker_host'),
+            'port' => config('services.micro.broker_port'),
+        ]);
+        $this->client = new EncodedConnection($options, $encoder);
+        $this->client->connect();
+    }
+
+    // 订阅消息
+    public function subscribe($topic, \Closure $callback)
+    {
+        $this->client->subscribe($topic, $callback);
+    }
+
+    // 发布消息
+    public function publish($topic, $message)
+    {
+        $this->client->publish($topic, $message);
+    }
+
+    // 同步请求
+    public function request($topic, $message, \Closure $callback)
+    {
+        $this->client->request($topic, $message, $callback);
+    }
+
+    // 等待消息
+    public function wait($number = 0)
+    {
+        $this->client->wait($number);
+    }
+}

composer.json

@@ -23,6 +23,7 @@
         "laravel/tinker": "~1.0",
         "nicolaslopezj/searchable": "^1.10",
         "paypal/rest-api-sdk-php": "*",
+        "repejota/nats": "^0.8.7",
         "santigarcor/laratrust": "5.0.*",
         "shippo/shippo-php": "^1.4"
     },

composer.lock

@@ -4,7 +4,7 @@
         "Read more about it at https://getcomposer.org/doc/01-basic-usage.md#installing-dependencies",
         "This file is @generated automatically"
     ],
-    "content-hash": "ce2fb9151a5b29afa4b1e2ed2f5c7579",
+    "content-hash": "26b8861e6f7bfbf1ccef3d6b9d9b651d",
     "packages": [
         {
             "name": "barryvdh/laravel-dompdf",
@@ -1332,6 +1332,115 @@
             },
             "time": "2020-09-30T07:37:11+00:00"
         },
+        {
+            "name": "ircmaxell/random-lib",
+            "version": "v1.2.0",
+            "source": {
+                "type": "git",
+                "url": "https://github.com/ircmaxell/RandomLib.git",
+                "reference": "e9e0204f40e49fa4419946c677eccd3fa25b8cf4"
+            },
+            "dist": {
+                "type": "zip",
+                "url": "https://api.github.com/repos/ircmaxell/RandomLib/zipball/e9e0204f40e49fa4419946c677eccd3fa25b8cf4",
+                "reference": "e9e0204f40e49fa4419946c677eccd3fa25b8cf4",
+                "shasum": ""
+            },
+            "require": {
+                "ircmaxell/security-lib": "^1.1",
+                "php": ">=5.3.2"
+            },
+            "require-dev": {
+                "friendsofphp/php-cs-fixer": "^1.11",
+                "mikey179/vfsstream": "^1.6",
+                "phpunit/phpunit": "^4.8|^5.0"
+            },
+            "type": "library",
+            "extra": {
+                "branch-alias": {
+                    "dev-master": "1.1.x-dev"
+                }
+            },
+            "autoload": {
+                "psr-0": {
+                    "RandomLib": "lib"
+                }
+            },
+            "notification-url": "https://packagist.org/downloads/",
+            "license": [
+                "MIT"
+            ],
+            "authors": [
+                {
+                    "name": "Anthony Ferrara",
+                    "email": "ircmaxell@ircmaxell.com",
+                    "homepage": "http://blog.ircmaxell.com"
+                }
+            ],
+            "description": "A Library For Generating Secure Random Numbers",
+            "homepage": "https://github.com/ircmaxell/RandomLib",
+            "keywords": [
+                "cryptography",
+                "random",
+                "random-numbers",
+                "random-strings"
+            ],
+            "support": {
+                "issues": "https://github.com/ircmaxell/RandomLib/issues",
+                "source": "https://github.com/ircmaxell/RandomLib/tree/master"
+            },
+            "time": "2016-09-07T15:52:06+00:00"
+        },
+        {
+            "name": "ircmaxell/security-lib",
+            "version": "v1.1.0",
+            "source": {
+                "type": "git",
+                "url": "https://github.com/ircmaxell/SecurityLib.git",
+                "reference": "f3db6de12c20c9bcd1aa3db4353a1bbe0e44e1b5"
+            },
+            "dist": {
+                "type": "zip",
+                "url": "https://api.github.com/repos/ircmaxell/SecurityLib/zipball/f3db6de12c20c9bcd1aa3db4353a1bbe0e44e1b5",
+                "reference": "f3db6de12c20c9bcd1aa3db4353a1bbe0e44e1b5",
+                "shasum": ""
+            },
+            "require": {
+                "php": ">=5.3.2"
+            },
+            "require-dev": {
+                "mikey179/vfsstream": "1.1.*"
+            },
+            "type": "library",
+            "extra": {
+                "branch-alias": {
+                    "dev-master": "1.0.x-dev"
+                }
+            },
+            "autoload": {
+                "psr-0": {
+                    "SecurityLib": "lib"
+                }
+            },
+            "notification-url": "https://packagist.org/downloads/",
+            "license": [
+                "MIT"
+            ],
+            "authors": [
+                {
+                    "name": "Anthony Ferrara",
+                    "email": "ircmaxell@ircmaxell.com",
+                    "homepage": "http://blog.ircmaxell.com"
+                }
+            ],
+            "description": "A Base Security Library",
+            "homepage": "https://github.com/ircmaxell/SecurityLib",
+            "support": {
+                "issues": "https://github.com/ircmaxell/SecurityLib/issues",
+                "source": "https://github.com/ircmaxell/SecurityLib/tree/master"
+            },
+            "time": "2015-03-20T14:31:23+00:00"
+        },
         {
             "name": "jakub-onderka/php-console-color",
             "version": "v0.2",
@@ -3779,6 +3888,72 @@
             },
             "time": "2020-02-21T04:36:14+00:00"
         },
+        {
+            "name": "repejota/nats",
+            "version": "0.8.7",
+            "source": {
+                "type": "git",
+                "url": "https://github.com/repejota/phpnats.git",
+                "reference": "68e6d24232e8b5aaf134697e47de2652b3ebff6f"
+            },
+            "dist": {
+                "type": "zip",
+                "url": "https://api.github.com/repos/repejota/phpnats/zipball/68e6d24232e8b5aaf134697e47de2652b3ebff6f",
+                "reference": "68e6d24232e8b5aaf134697e47de2652b3ebff6f",
+                "shasum": ""
+            },
+            "require": {
+                "ircmaxell/random-lib": "^1.2"
+            },
+            "require-dev": {
+                "leanphp/phpspec-code-coverage": "^3.2@dev",
+                "phpspec/phpspec": "^3.0",
+                "phpunit/phpunit": "5.*",
+                "satooshi/php-coveralls": "2.0.x-dev"
+            },
+            "type": "library",
+            "autoload": {
+                "psr-0": {
+                    "Nats": "src",
+                    "Nats\\Test": "src"
+                }
+            },
+            "notification-url": "https://packagist.org/downloads/",
+            "license": [
+                "MIT"
+            ],
+            "authors": [
+                {
+                    "name": "Raül Pérez",
+                    "email": "repejota@gmail.com",
+                    "homepage": "http://www.repejota.com",
+                    "role": "Developer"
+                },
+                {
+                    "name": "Adrià Cidre",
+                    "email": "adria.cidre@gmail.com",
+                    "homepage": "http://oridoki.com",
+                    "role": "Developer"
+                },
+                {
+                    "name": "José Gil",
+                    "email": "josgilmo@gmail.com",
+                    "role": "Developer"
+                },
+                {
+                    "name": "Gorka López de Torre",
+                    "email": "glopezdetorre@gmail.com",
+                    "homepage": "http://gorka.io",
+                    "role": "Developer"
+                }
+            ],
+            "description": "A nats.io client in PHP",
+            "support": {
+                "issues": "https://github.com/repejota/phpnats/issues",
+                "source": "https://github.com/repejota/phpnats/tree/master"
+            },
+            "time": "2018-06-09T15:32:07+00:00"
+        },
         {
             "name": "sabberworm/php-css-parser",
             "version": "8.3.1",

config/services.php

@@ -40,6 +40,8 @@ return [
         'timeout' => env('MICRO_TIMEOUT', 3.0),   //网关超时时间
         'jwt_key' => env('MICRO_JWT_KEY', 'laracomUserTokenKeySecret'),
         'jwt_algorithms' => env('MICRO_JWT_ALGORITHMS', 'HS256'),
+        'broker_host' => env('MICRO_BROKER_HOST', '127.0.0.1'),
+        'broker_port' => env('MICRO_BROKER_PORT', '4222'),
     ]
 
 ];