Commit f19360e9 by banice

修改topic共用同一个

parent a68a3ab6
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/10/11
* Time: 上午10:49
*/
namespace App\Http\Controllers;
use Illuminate\Http\Request;
class TestController extends Controller
{
public function test()
{
}
private function hello(Request $request)
{
return $request->all();
}
public function testDirection(Request $request)
{
dd($request->all());
}
}
\ No newline at end of file
...@@ -9,47 +9,128 @@ ...@@ -9,47 +9,128 @@
namespace App\Http\Controllers; namespace App\Http\Controllers;
use App\Http\model\BatteryModel;
use App\model\OrderModel;
use Illuminate\Http\Request; use Illuminate\Http\Request;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log; use Illuminate\Support\Facades\Log;
use Lib\aliyun\api\PublishToTopic; use Lib\aliyun\api\PublishToTopic;
use Lib\aliyun\Client; use Lib\aliyun\Client;
use Lib\command\ConfirmOpenDoor; use Lib\command\ConfirmOpenDoor;
use Lib\command\OpenDoor;
use phpDocumentor\Reflection\Types\Self_;
use Src\battery\Battery;
use Src\box\Box; use Src\box\Box;
use Src\box\door\Door;
use Src\order\Order;
use Src\order\OrderCache;
use Src\pay_rule\CashPay;
class UpdateController extends Controller class UpdateController extends Controller
{ {
public function update(Request $request) const COMMAND = [
'return_open_door' => 11,
'confirm_return_open_door' => 92,
'close_return_door' => 93,
];
public function update(Request $request, Order $order)
{ {
Log::debug('update'); $payload = $request->get('payload');
Log::debug($request->all()); return;
$payload = $request->get('payload'); $client = $payload['client'];
$door = $payload['doors'];
$boxNum = $door['box_num'];
$doorNum = $door['door_num'];
$operation = $door['operation'];
$orderId = $door['order_id'];
$boxCode = $request->get('box_code');
//todo 检查订单是否存在,websocket推送打开门成功,检查电池是否放入,
$confirmOpenDoor = new ConfirmOpenDoor();
$confirmOpenDoor->setOrderId($orderId);
$publishToTopic = new PublishToTopic();
$publishToTopic->setMessageContent($confirmOpenDoor->getMessage());
$publishToTopic->setProductKey(env('MQTT_PRODUCT_KEY'));
$publishToTopic->setTopicFullName(Box::getOpenDoorTopic($boxCode));
$client = new Client();
$client->setApi($publishToTopic);
if ($client->request() === true) {
Log::debug('confirm order -' . $orderId);
if ($client === 'battery') {
Log::debug('battery');
return;
} }
//hdg
$topic = $payload['hdg_topic'];
switch ($topic) {
case 'open_door':
break;
default:
break;
} }
switch ($payload['command']) {
case self::COMMAND['return_open_door']:
break;
case self::COMMAND['close_return_door']:
break;
case 0:
break;
default:
break;
}
//todo 根据订单号查找到用户的电池比对,比对打开舱门是否正确
break;
default:
break;
}
}
} }
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/10/12
* Time: 上午12:43
*/
namespace App\Http\Controllers\battery;
use App\Http\Controllers\Controller;
use Illuminate\Http\Request;
use Illuminate\Support\Facades\Log;
class UpdateController extends Controller
{
public function update(Request $request)
{
Log::debug($request->all());
}
}
\ No newline at end of file
...@@ -11,7 +11,7 @@ namespace App\Http\Controllers\door; ...@@ -11,7 +11,7 @@ namespace App\Http\Controllers\door;
use App\Http\Controllers\Controller; use App\Http\Controllers\Controller;
use Illuminate\Http\Request; use Illuminate\Http\Request;
use src\box\Box; use Src\box\Box;
use Validator; use Validator;
class InitController extends Controller class InitController extends Controller
......
...@@ -19,12 +19,14 @@ class StateController extends Controller ...@@ -19,12 +19,14 @@ class StateController extends Controller
public function state(Request $request) public function state(Request $request)
{ {
Log::debug('door');
Log::debug('haha');
$boxCode = $request->get('box_code'); $boxCode = $request->get('box_code');
$payload = $request->get('payload'); $payload = $request->get('payload');
Log::debug($payload);
$validator = Validator::make($payload, [ $validator = Validator::make($payload, [
'door_count' => 'required|integer', 'door_count' => 'required|integer',
'door' => 'required|array' 'door' => 'required|array'
...@@ -33,14 +35,9 @@ class StateController extends Controller ...@@ -33,14 +35,9 @@ class StateController extends Controller
$doorCount = $payload['door_count']; $doorCount = $payload['door_count'];
$doorArray = $payload['door']; $doorArray = $payload['door'];
if (count($doorArray) === (int)$doorCount) { // if (count($doorArray) === (int)$doorCount) {
//
foreach ($doorArray as $item) { // }
Log::debug($item);
}
}
} }
} }
\ No newline at end of file
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/10/11
* Time: 上午9:55
*/
namespace App\Http\Middleware;
use Illuminate\Http\Request;
use Closure;
use Illuminate\Support\Facades\Log;
use Lib\messageId\MessageExist;
class MessageLogMiddleware
{
/**
* @param Request $request
* @param Closure $next
* @return mixed
*/
public function handle(Request $request, Closure $next)
{
$response = $next($request);
$state = $request->get('message_state', MessageExist::STATE['handle']);
$messageId = $request->get('message_id');
$messageExist = new MessageExist();
$messageExist->add($messageId, $state);
return $response;
}
}
\ No newline at end of file
...@@ -14,6 +14,7 @@ use Closure; ...@@ -14,6 +14,7 @@ use Closure;
use Illuminate\Http\Request; use Illuminate\Http\Request;
use Illuminate\Http\Response; use Illuminate\Http\Response;
use Illuminate\Support\Facades\Log; use Illuminate\Support\Facades\Log;
use Lib\messageId\MessageExist;
use Validator; use Validator;
class MqttMiddleware class MqttMiddleware
...@@ -21,9 +22,7 @@ class MqttMiddleware ...@@ -21,9 +22,7 @@ class MqttMiddleware
public function handle(Request $request, Closure $next) public function handle(Request $request, Closure $next)
{ {
$data = $request->all(); Log::debug($request->all());
Log::debug('mqtt middleware');
$validator = Validator::make($request->all(), [ $validator = Validator::make($request->all(), [
...@@ -39,28 +38,39 @@ class MqttMiddleware ...@@ -39,28 +38,39 @@ class MqttMiddleware
if ($validator->fails() === true) { if ($validator->fails() === true) {
Log::error($validator->errors()->first()); Log::error($validator->errors()->first());
return; return new Response();
abort(400, $validator->errors()->first());
} }
if (strtoupper(md5($request->get('Message'))) != $request->get('MessageMD5')) { if (strtoupper(md5($request->get('Message'))) != $request->get('MessageMD5')) {
Log::error('md5 error' . $request->get('MessageId')); Log::error('md5 error' . $request->get('MessageId'));
abort(400, ' invalid ');
return new Response(); return new Response();
} }
$message = json_decode($request->get('Message'), true); $message = json_decode($request->get('Message'), true);
$messageId = $message['MessageId'];
app()->register(RedisServiceProvider::class); Log::debug($message);
$messageId = $request->get('MessageId');
$messageExist = new MessageExist();
try {
if ($messageExist->exists($messageId) === true) {
Log::debug('message existed');
return new Response();
}
} catch (\Exception $exception) {
Log::error($exception->getMessage());
}
$boxCode = $message['box_code']; $boxCode = $message['box_code'];
$payload = json_decode($message['payload'], true); $payload = json_decode($message['payload'], true);
$request->merge(['box_code' => $boxCode, 'payload' => $payload]); $request->merge(['box_code' => $boxCode, 'payload' => $payload, 'message_id' => $messageId]);
return $next($request); return $next($request);
......
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/10/14
* Time: 下午8:51
*/
namespace App\Http\model;
use Illuminate\Database\Eloquent\Model;
class BatteryModel extends Model
{
protected $table = 'batteries';
}
\ No newline at end of file
...@@ -22,9 +22,7 @@ class RedisServiceProvider extends ServiceProvider ...@@ -22,9 +22,7 @@ class RedisServiceProvider extends ServiceProvider
$this->app->singleton('Redis', function ($app) { $this->app->singleton('Redis', function ($app) {
$redis = new \Redis(); $redis = new \Redis();
$config = config('database.redis.default'); $config = config('database.redis.default');
$redis->connect($config['host']); $redis->connect($config['host']);
$redis->select($config['database']); $redis->select($config['database']);
......
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/9/13
* Time: 下午7:29
*/
namespace App\model;
use Illuminate\Database\Eloquent\Model;
class ActivityModel extends Model
{
protected $table = 'activities';
}
\ No newline at end of file
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/9/17
* Time: 下午8:40
*/
namespace App\model;
use Illuminate\Database\Eloquent\Model;
class BoxModel extends Model
{
protected $table = 'boxes';
}
\ No newline at end of file
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/10/8
* Time: 上午10:43
*/
namespace App\model;
use Illuminate\Database\Eloquent\Model;
use Src\box\door\Door;
class DoorModel extends Model
{
protected $table = 'box_rooms';
public function add(Door $door)
{
$this->name = $door->getName();
$this->box_type = $door->getBoxType();
$this->box_id = $door->getBoxId();
return $this->save();
}
}
\ No newline at end of file
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/10/7
* Time: 下午9:22
*/
namespace App\model;
use Illuminate\Database\Eloquent\Model;
use Src\order\Order;
use DB;
class OrderModel extends Model
{
protected $table = 'orders';
public function initOrder(Order $order)
{
$this->code = $order->getOutTradeNo();
$this->guest_id = $order->getUserId();
$this->site_id = $order->getSite()->getId();
$this->site_name = $order->getSite()->getName();
$this->out_room_id = $order->getTakeDoor()->getId();
$this->out_box_id = $order->getTakeBox()->getId();
$this->out_room_name = $order->getTakeDoor()->getName();
$this->box_code = $order->getBoxCode();
$this->status = 1;
$this->out_date = date('Y-m-d H:i:s');
return $this->save();
}
public function searchLastOrder(int $userId)
{
return DB::select('select * from orders where id = (select max(id) from orders where guest_id = ' . $userId . ')');
}
}
\ No newline at end of file
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/9/13
* Time: 下午7:10
*/
namespace App\model;
use Illuminate\Database\Eloquent\Model;
class PayGroupModel extends Model
{
protected $table = 'pay_groups';
}
\ No newline at end of file
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/9/13
* Time: 下午2:28
*/
namespace App\model;
use Illuminate\Database\Eloquent\Model;
class SystemModel extends Model
{
protected $table = 'system';
}
\ No newline at end of file
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/9/13
* Time: 下午7:34
*/
namespace App\model;
use Illuminate\Database\Eloquent\Model;
use DB;
class TicketModel extends Model
{
protected $table = 'tickets';
/**
* @param int $userId
* @return int
*/
public function getUserTicket(int $userId): int
{
$today = date('Y-m-d H:i:s');
$ticketId = DB::table('tickets')->where([
['begin_date', '<', $today],
['end_date', '>', $today],
['guest_id', $userId],
['status', 1],
])->min('id');
if (blank($ticketId)) {
return 0;
}
$amount = DB::table('tickets')->where('id', $ticketId)->first(['amount']);
return $amount['amount'];
}
}
\ No newline at end of file
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/9/7
* Time: 上午11:37
*/
namespace App\model;
use Illuminate\Database\Eloquent\Model;
use Lib\wx\User;
use Tymon\JWTAuth\Contracts\JWTSubject;
use Illuminate\Auth\Authenticatable;
use Illuminate\Support\Facades\Log;
use Laravel\Lumen\Auth\Authorizable;
use Illuminate\Contracts\Auth\Authenticatable as AuthenticatableContract;
use Illuminate\Contracts\Auth\Access\Authorizable as AuthorizableContract;
class UserModel extends Model implements JWTSubject, AuthenticatableContract, AuthorizableContract
{
use Authenticatable, Authorizable;
protected $table = 'guests';
public function getJWTIdentifier()
{
return $this->getKey();
}
public function getJWTCustomClaims()
{
$return = [
'wx_authorize' => false
];
if (isset($this->open_id) === true) {
if ($this->open_id != null) {
$return['wx_authorize'] = true;
}
}
return $return;
}
public function addWxUser(User $user)
{
$this->open_id = $user->getOpenId();
$this->session_key = $user->getSessionId();
$this->telephone = $user->getPhone();
$this->pure_phone = $user->getPurePhone();
$this->country_code = $user->getCountryCode();
return $this->save();
}
}
\ No newline at end of file
...@@ -72,7 +72,8 @@ $app->singleton( ...@@ -72,7 +72,8 @@ $app->singleton(
$app->routeMiddleware([ $app->routeMiddleware([
// 'auth' => App\Http\Middleware\Authenticate::class, // 'auth' => App\Http\Middleware\Authenticate::class,
'mqtt' => \App\Http\Middleware\MqttMiddleware::class, 'mqtt' => \App\Http\Middleware\MqttMiddleware::class,
'message_exist' => \App\Http\Middleware\MessageLogMiddleware::class,
]); ]);
/* /*
...@@ -87,7 +88,7 @@ $app->routeMiddleware([ ...@@ -87,7 +88,7 @@ $app->routeMiddleware([
*/ */
$app->register(App\Providers\AppServiceProvider::class); $app->register(App\Providers\AppServiceProvider::class);
//$app->register(\App\Providers\RedisServiceProvider::class); $app->register(\App\Providers\RedisServiceProvider::class);
// $app->register(App\Providers\AuthServiceProvider::class); // $app->register(App\Providers\AuthServiceProvider::class);
// $app->register(App\Providers\EventServiceProvider::class); // $app->register(App\Providers\EventServiceProvider::class);
......
...@@ -11,5 +11,6 @@ return [ ...@@ -11,5 +11,6 @@ return [
'AC_ALARM' => 'hdg:ac:alarm:hash', 'AC_ALARM' => 'hdg:ac:alarm:hash',
'MESSAGE_EXIST' => 'hdg:message:exist', 'MESSAGE_EXIST' => 'hdg:message:exist',
'BATTERY_QUANTITY' => 'hdg:battery:quantity:',
]; ];
<?php
namespace Lib;
use \Exception;
/**
* This file is part of workerman.
*
* Licensed under The MIT License
* For full copyright and license information, please see the MIT-LICENSE.txt
* Redistributions of files must retain the above copyright notice.
*
* @author walkor<walkor@workerman.net>
* @copyright walkor<walkor@workerman.net>
* @link http://www.workerman.net/
* @license http://www.opensource.org/licenses/mit-license.php MIT License
*/
/**
* 数据发送相关
* @version 3.0.10
*/
class Gateway
{
/**
* gateway 实例
*
* @var object
*/
protected static $businessWorker = null;
/**
* 注册中心地址
*
* @var string
*/
// public static $registerAddress = 'cdzregister.default.svc.cluster.local:1239';
public static $registerAddress = '100.103.0.9:1239';
//public static $registerAddress = '127.0.0.1:1236';
/**
* 秘钥
* @var string
*/
public static $secretKey = '';
/**
* 链接超时时间
* @var int
*/
public static $connectTimeout = 3;
/**
* 与Gateway是否是长链接
* @var bool
*/
public static $persistentConnection = false;
/**
* 向所有客户端连接(或者 client_id_array 指定的客户端连接)广播消息
*
* @param string $message 向客户端发送的消息
* @param array $client_id_array 客户端 id 数组
* @param array $exclude_client_id 不给这些client_id发
* @param bool $raw 是否发送原始数据(即不调用gateway的协议的encode方法)
* @return void
* @throws Exception
*/
public static function sendToAll($message, $client_id_array = null, $exclude_client_id = null, $raw = false)
{
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_SEND_TO_ALL;
$gateway_data['body'] = $message;
if ($raw) {
$gateway_data['flag'] |= GatewayProtocol::FLAG_NOT_CALL_ENCODE;
}
if ($exclude_client_id) {
if (!is_array($exclude_client_id)) {
$exclude_client_id = array($exclude_client_id);
}
if ($client_id_array) {
$exclude_client_id = array_flip($exclude_client_id);
}
}
if ($client_id_array) {
if (!is_array($client_id_array)) {
echo new \Exception('bad $client_id_array:' . var_export($client_id_array, true));
return;
}
$data_array = array();
foreach ($client_id_array as $client_id) {
if (isset($exclude_client_id[$client_id])) {
continue;
}
$address = Context::clientIdToAddress($client_id);
if ($address) {
$key = long2ip($address['local_ip']) . ":{$address['local_port']}";
$data_array[$key][$address['connection_id']] = $address['connection_id'];
}
}
foreach ($data_array as $addr => $connection_id_list) {
$the_gateway_data = $gateway_data;
$the_gateway_data['ext_data'] = json_encode(array('connections' => $connection_id_list));
static::sendToGateway($addr, $the_gateway_data);
}
return;
} elseif (empty($client_id_array) && is_array($client_id_array)) {
return;
}
if (!$exclude_client_id) {
return static::sendToAllGateway($gateway_data);
}
$address_connection_array = static::clientIdArrayToAddressArray($exclude_client_id);
// 如果有businessWorker实例,说明运行在workerman环境中,通过businessWorker中的长连接发送数据
if (static::$businessWorker) {
foreach (static::$businessWorker->gatewayConnections as $address => $gateway_connection) {
$gateway_data['ext_data'] = isset($address_connection_array[$address]) ?
json_encode(array('exclude' => $address_connection_array[$address])) : '';
/** @var TcpConnection $gateway_connection */
$gateway_connection->send($gateway_data);
}
} // 运行在其它环境中,通过注册中心得到gateway地址
else {
$all_addresses = static::getAllGatewayAddressesFromRegister();
if (!$all_addresses) {
throw new Exception('Gateway::getAllGatewayAddressesFromRegister() with registerAddress:' .
static::$registerAddress . ' return ' . var_export($all_addresses, true));
}
foreach ($all_addresses as $address) {
$gateway_data['ext_data'] = isset($address_connection_array[$address]) ?
json_encode(array('exclude' => $address_connection_array[$address])) : '';
static::sendToGateway($address, $gateway_data);
}
}
}
/**
* 向某个client_id对应的连接发消息
*
* @param int $client_id
* @param string $message
* @return void
*/
public static function sendToClient($client_id, $message)
{
return static::sendCmdAndMessageToClient($client_id, GatewayProtocol::CMD_SEND_TO_ONE, $message);
}
/**
* 判断某个uid是否在线
*
* @param string $uid
* @return int 0|1
*/
public static function isUidOnline($uid)
{
return (int)static::getClientIdByUid($uid);
}
/**
* 判断client_id对应的连接是否在线
*
* @param int $client_id
* @return int 0|1
*/
public static function isOnline($client_id)
{
$address_data = Context::clientIdToAddress($client_id);
if (!$address_data) {
return 0;
}
$address = long2ip($address_data['local_ip']) . ":{$address_data['local_port']}";
if (isset(static::$businessWorker)) {
if (!isset(static::$businessWorker->gatewayConnections[$address])) {
return 0;
}
}
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_IS_ONLINE;
$gateway_data['connection_id'] = $address_data['connection_id'];
return (int)static::sendAndRecv($address, $gateway_data);
}
/**
* 获取所有在线用户的session,client_id为 key(弃用,请用getAllClientSessions代替)
*
* @param string $group
* @return array
*/
public static function getAllClientInfo($group = '')
{
echo "Warning: Gateway::getAllClientInfo is deprecated and will be removed in a future, please use Gateway::getAllClientSessions instead.";
return static::getAllClientSessions($group);
}
/**
* 获取所有在线client_id的session,client_id为 key
*
* @param string $group
* @return array
*/
public static function getAllClientSessions($group = '')
{
$gateway_data = GatewayProtocol::$empty;
if (!$group) {
$gateway_data['cmd'] = GatewayProtocol::CMD_GET_ALL_CLIENT_SESSIONS;
} else {
$gateway_data['cmd'] = GatewayProtocol::CMD_GET_CLINET_SESSUONS_BY_GROUP;
$gateway_data['ext_data'] = $group;
}
$status_data = array();
$all_buffer_array = static::getBufferFromAllGateway($gateway_data);
foreach ($all_buffer_array as $local_ip => $buffer_array) {
foreach ($buffer_array as $local_port => $data) {
if ($data) {
foreach ($data as $connection_id => $session_buffer) {
$client_id = Context::addressToClientId($local_ip, $local_port, $connection_id);
if ($client_id === Context::$client_id) {
$status_data[$client_id] = (array)$_SESSION;
} else {
$status_data[$client_id] = $session_buffer ? Context::sessionDecode($session_buffer) : array();
}
}
}
}
}
return $status_data;
}
/**
* 获取某个组的连接信息(弃用,请用getClientSessionsByGroup代替)
*
* @param string $group
* @return array
*/
public static function getClientInfoByGroup($group)
{
echo "Warning: Gateway::getClientInfoByGroup is deprecated and will be removed in a future, please use Gateway::getClientSessionsByGroup instead.";
return static::getAllClientSessions($group);
}
/**
* 获取某个组的所有client_id的session信息
*
* @param string $group
*
* @return array
*/
public static function getClientSessionsByGroup($group)
{
if (static::isValidGroupId($group)) {
return static::getAllClientSessions($group);
}
return array();
}
/**
* 获取所有在线client_id数
*
* @return int
*/
public static function getAllClientIdCount()
{
return static::getClientCountByGroup();
}
/**
* 获取所有在线client_id数(getAllClientIdCount的别名)
*
* @return int
*/
public static function getAllClientCount()
{
return static::getAllClientIdCount();
}
/**
* 获取某个组的在线client_id数
*
* @param string $group
* @return int
*/
public static function getClientIdCountByGroup($group = '')
{
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_GET_CLIENT_COUNT_BY_GROUP;
$gateway_data['ext_data'] = $group;
$total_count = 0;
$all_buffer_array = static::getBufferFromAllGateway($gateway_data);
foreach ($all_buffer_array as $local_ip => $buffer_array) {
foreach ($buffer_array as $local_port => $count) {
if ($count) {
$total_count += $count;
}
}
}
return $total_count;
}
/**
* getClientIdCountByGroup 函数的别名
*
* @param string $group
* @return int
*/
public static function getClientCountByGroup($group = '')
{
return static::getClientIdCountByGroup($group);
}
/**
* 获取某个群组在线client_id列表
*
* @param string $group
* @return array
*/
public static function getClientIdListByGroup($group)
{
if (!static::isValidGroupId($group)) {
return array();
}
$data = static::select(array('uid'), array('groups' => is_array($group) ? $group : array($group)));
$client_id_map = array();
foreach ($data as $local_ip => $buffer_array) {
foreach ($buffer_array as $local_port => $items) {
//$items = ['connection_id'=>['uid'=>x, 'group'=>[x,x..], 'session'=>[..]], 'client_id'=>[..], ..];
foreach ($items as $connection_id => $info) {
$client_id = Context::addressToClientId($local_ip, $local_port, $connection_id);
$client_id_map[$client_id] = $client_id;
}
}
}
return $client_id_map;
}
/**
* 获取集群所有在线client_id列表
*
* @return array
*/
public static function getAllClientIdList()
{
return static::formatClientIdFromGatewayBuffer(static::select(array('uid')));
}
/**
* 格式化client_id
*
* @param $data
* @return array
*/
protected static function formatClientIdFromGatewayBuffer($data)
{
$client_id_list = array();
foreach ($data as $local_ip => $buffer_array) {
foreach ($buffer_array as $local_port => $items) {
//$items = ['connection_id'=>['uid'=>x, 'group'=>[x,x..], 'session'=>[..]], 'client_id'=>[..], ..];
foreach ($items as $connection_id => $info) {
$client_id = Context::addressToClientId($local_ip, $local_port, $connection_id);
$client_id_list[$client_id] = $client_id;
}
}
}
return $client_id_list;
}
/**
* 获取与 uid 绑定的 client_id 列表
*
* @param string $uid
* @return array
*/
public static function getClientIdByUid($uid)
{
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_GET_CLIENT_ID_BY_UID;
$gateway_data['ext_data'] = $uid;
$client_list = array();
$all_buffer_array = static::getBufferFromAllGateway($gateway_data);
foreach ($all_buffer_array as $local_ip => $buffer_array) {
foreach ($buffer_array as $local_port => $connection_id_array) {
if ($connection_id_array) {
foreach ($connection_id_array as $connection_id) {
$client_list[] = Context::addressToClientId($local_ip, $local_port, $connection_id);
}
}
}
}
return $client_list;
}
/**
* 获取某个群组在线uid列表
*
* @param string $group
* @return array
*/
public static function getUidListByGroup($group)
{
if (!static::isValidGroupId($group)) {
return array();
}
$group = is_array($group) ? $group : array($group);
$data = static::select(array('uid'), array('groups' => $group));
$uid_map = array();
foreach ($data as $local_ip => $buffer_array) {
foreach ($buffer_array as $local_port => $items) {
//$items = ['connection_id'=>['uid'=>x, 'group'=>[x,x..], 'session'=>[..]], 'client_id'=>[..], ..];
foreach ($items as $connection_id => $info) {
if (!empty($info['uid'])) {
$uid_map[$info['uid']] = $info['uid'];
}
}
}
}
return $uid_map;
}
/**
* 获取某个群组在线uid数
*
* @param string $group
* @return int
*/
public static function getUidCountByGroup($group)
{
if (static::isValidGroupId($group)) {
return count(static::getUidListByGroup($group));
}
return 0;
}
/**
* 获取全局在线uid列表
*
* @return array
*/
public static function getAllUidList()
{
$data = static::select(array('uid'));
$uid_map = array();
foreach ($data as $local_ip => $buffer_array) {
foreach ($buffer_array as $local_port => $items) {
//$items = ['connection_id'=>['uid'=>x, 'group'=>[x,x..], 'session'=>[..]], 'client_id'=>[..], ..];
foreach ($items as $connection_id => $info) {
if (!empty($info['uid'])) {
$uid_map[$info['uid']] = $info['uid'];
}
}
}
}
return $uid_map;
}
/**
* 获取全局在线uid数
* @return int
*/
public static function getAllUidCount()
{
return count(static::getAllUidList());
}
/**
* 通过client_id获取uid
*
* @param $client_id
* @return mixed
*/
public static function getUidByClientId($client_id)
{
$data = static::select(array('uid'), array('client_id' => array($client_id)));
foreach ($data as $local_ip => $buffer_array) {
foreach ($buffer_array as $local_port => $items) {
//$items = ['connection_id'=>['uid'=>x, 'group'=>[x,x..], 'session'=>[..]], 'client_id'=>[..], ..];
foreach ($items as $info) {
return $info['uid'];
}
}
}
}
/**
* 获取所有在线的群组id
*
* @return array
*/
public static function getAllGroupIdList()
{
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_GET_GROUP_ID_LIST;
$group_id_list = array();
$all_buffer_array = static::getBufferFromAllGateway($gateway_data);
foreach ($all_buffer_array as $local_ip => $buffer_array) {
foreach ($buffer_array as $local_port => $group_id_array) {
if (is_array($group_id_array)) {
foreach ($group_id_array as $group_id) {
if (!isset($group_id_list[$group_id])) {
$group_id_list[$group_id] = $group_id;
}
}
}
}
}
return $group_id_list;
}
/**
* 获取所有在线分组的uid数量,也就是每个分组的在线用户数
*
* @return array
*/
public static function getAllGroupUidCount()
{
$group_uid_map = static::getAllGroupUidList();
$group_uid_count_map = array();
foreach ($group_uid_map as $group_id => $uid_list) {
$group_uid_count_map[$group_id] = count($uid_list);
}
return $group_uid_count_map;
}
/**
* 获取所有分组uid在线列表
*
* @return array
*/
public static function getAllGroupUidList()
{
$data = static::select(array('uid', 'groups'));
$group_uid_map = array();
foreach ($data as $local_ip => $buffer_array) {
foreach ($buffer_array as $local_port => $items) {
//$items = ['connection_id'=>['uid'=>x, 'group'=>[x,x..], 'session'=>[..]], 'client_id'=>[..], ..];
foreach ($items as $connection_id => $info) {
if (empty($info['uid']) || empty($info['groups'])) {
break;
}
$uid = $info['uid'];
foreach ($info['groups'] as $group_id) {
if (!isset($group_uid_map[$group_id])) {
$group_uid_map[$group_id] = array();
}
$group_uid_map[$group_id][$uid] = $uid;
}
}
}
}
return $group_uid_map;
}
/**
* 获取所有群组在线client_id列表
*
* @return array
*/
public static function getAllGroupClientIdList()
{
$data = static::select(array('groups'));
$group_client_id_map = array();
foreach ($data as $local_ip => $buffer_array) {
foreach ($buffer_array as $local_port => $items) {
//$items = ['connection_id'=>['uid'=>x, 'group'=>[x,x..], 'session'=>[..]], 'client_id'=>[..], ..];
foreach ($items as $connection_id => $info) {
if (empty($info['groups'])) {
break;
}
$client_id = Context::addressToClientId($local_ip, $local_port, $connection_id);
foreach ($info['groups'] as $group_id) {
if (!isset($group_client_id_map[$group_id])) {
$group_client_id_map[$group_id] = array();
}
$group_client_id_map[$group_id][$client_id] = $client_id;
}
}
}
}
return $group_client_id_map;
}
/**
* 获取所有群组在线client_id数量,也就是获取每个群组在线连接数
*
* @return array
*/
public static function getAllGroupClientIdCount()
{
$group_client_map = static::getAllGroupClientIdList();
$group_client_count_map = array();
foreach ($group_client_map as $group_id => $client_id_list) {
$group_client_count_map[$group_id] = count($client_id_list);
}
return $group_client_count_map;
}
/**
* 根据条件到gateway搜索数据
*
* @param array $fields
* @param array $where
* @return array
*/
protected static function select($fields = array('session', 'uid', 'groups'), $where = array())
{
$t = microtime(true);
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_SELECT;
$gateway_data['ext_data'] = array('fields' => $fields, 'where' => $where);
$gateway_data_list = array();
// 有client_id,能计算出需要和哪些gateway通讯,只和必要的gateway通讯能降低系统负载
if (isset($where['client_id'])) {
$client_id_list = $where['client_id'];
unset($gateway_data['ext_data']['where']['client_id']);
$gateway_data['ext_data']['where']['connection_id'] = array();
foreach ($client_id_list as $client_id) {
$address_data = Context::clientIdToAddress($client_id);
if (!$address_data) {
continue;
}
$address = long2ip($address_data['local_ip']) . ":{$address_data['local_port']}";
if (!isset($gateway_data_list[$address])) {
$gateway_data_list[$address] = $gateway_data;
}
$gateway_data_list[$address]['ext_data']['where']['connection_id'][$address_data['connection_id']] = $address_data['connection_id'];
}
foreach ($gateway_data_list as $address => $item) {
$gateway_data_list[$address]['ext_data'] = json_encode($item['ext_data']);
}
// 有其它条件,则还是需要向所有gateway发送
if (count($where) !== 1) {
$gateway_data['ext_data'] = json_encode($gateway_data['ext_data']);
foreach (static::getAllGatewayAddress() as $address) {
if (!isset($gateway_data_list[$address])) {
$gateway_data_list[$address] = $gateway_data;
}
}
}
$data = static::getBufferFromSomeGateway($gateway_data_list);
} else {
$gateway_data['ext_data'] = json_encode($gateway_data['ext_data']);
$data = static::getBufferFromAllGateway($gateway_data);
}
return $data;
}
/**
* 生成验证包,用于验证此客户端的合法性
*
* @return string
*/
protected static function generateAuthBuffer()
{
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_GATEWAY_CLIENT_CONNECT;
$gateway_data['body'] = json_encode(array(
'secret_key' => static::$secretKey,
));
return GatewayProtocol::encode($gateway_data);
}
/**
* 批量向某些gateway发包,并得到返回数组
*
* @param array $gateway_data_array
* @return array
* @throws Exception
*/
protected static function getBufferFromSomeGateway($gateway_data_array)
{
$gateway_buffer_array = array();
$auth_buffer = static::$secretKey ? static::generateAuthBuffer() : '';
foreach ($gateway_data_array as $address => $gateway_data) {
if ($auth_buffer) {
$gateway_buffer_array[$address] = $auth_buffer . GatewayProtocol::encode($gateway_data);
} else {
$gateway_buffer_array[$address] = GatewayProtocol::encode($gateway_data);
}
}
return static::getBufferFromGateway($gateway_buffer_array);
}
/**
* 批量向所有 gateway 发包,并得到返回数组
*
* @param string $gateway_data
* @return array
* @throws Exception
*/
protected static function getBufferFromAllGateway($gateway_data)
{
$addresses = static::getAllGatewayAddress();
$gateway_buffer_array = array();
$gateway_buffer = GatewayProtocol::encode($gateway_data);
if (!is_array($gateway_data)) {
$gateway_buffer = static::$secretKey ? static::generateAuthBuffer() . $gateway_buffer : $gateway_buffer;
}
foreach ($addresses as $address) {
$gateway_buffer_array[$address] = $gateway_buffer;
}
return static::getBufferFromGateway($gateway_buffer_array);
}
/**
* 获取所有gateway内部通讯地址
*
* @return array
* @throws Exception
*/
protected static function getAllGatewayAddress()
{
if (isset(static::$businessWorker)) {
$addresses = static::$businessWorker->getAllGatewayAddresses();
if (empty($addresses)) {
throw new Exception('businessWorker::getAllGatewayAddresses return empty');
}
} else {
$addresses = static::getAllGatewayAddressesFromRegister();
if (empty($addresses)) {
return array();
}
}
return $addresses;
}
/**
* 批量向gateway发送并获取数据
* @param $gateway_buffer_array
* @return array
*/
protected static function getBufferFromGateway($gateway_buffer_array)
{
$client_array = $status_data = $client_address_map = $receive_buffer_array = $recv_length_array = array();
// 批量向所有gateway进程发送请求数据
foreach ($gateway_buffer_array as $address => $gateway_buffer) {
$client = stream_socket_client("tcp://$address", $errno, $errmsg, static::$connectTimeout);
if ($client && strlen($gateway_buffer) === stream_socket_sendto($client, $gateway_buffer)) {
$socket_id = (int)$client;
$client_array[$socket_id] = $client;
$client_address_map[$socket_id] = explode(':', $address);
$receive_buffer_array[$socket_id] = '';
}
}
// 超时5秒
$timeout = 5;
$time_start = microtime(true);
// 批量接收请求
while (count($client_array) > 0) {
$write = $except = array();
$read = $client_array;
if (@stream_select($read, $write, $except, $timeout)) {
foreach ($read as $client) {
$socket_id = (int)$client;
$buffer = stream_socket_recvfrom($client, 65535);
if ($buffer !== '' && $buffer !== false) {
$receive_buffer_array[$socket_id] .= $buffer;
$receive_length = strlen($receive_buffer_array[$socket_id]);
if (empty($recv_length_array[$socket_id]) && $receive_length >= 4) {
$recv_length_array[$socket_id] = current(unpack('N', $receive_buffer_array[$socket_id]));
}
if (!empty($recv_length_array[$socket_id]) && $receive_length >= $recv_length_array[$socket_id] + 4) {
unset($client_array[$socket_id]);
}
} elseif (feof($client)) {
unset($client_array[$socket_id]);
}
}
}
if (microtime(true) - $time_start > $timeout) {
break;
}
}
$format_buffer_array = array();
foreach ($receive_buffer_array as $socket_id => $buffer) {
$local_ip = ip2long($client_address_map[$socket_id][0]);
$local_port = $client_address_map[$socket_id][1];
$format_buffer_array[$local_ip][$local_port] = unserialize(substr($buffer, 4));
}
return $format_buffer_array;
}
/**
* 踢掉某个客户端,并以$message通知被踢掉客户端
*
* @param int $client_id
* @param string $message
* @return void
*/
public static function closeClient($client_id, $message = null)
{
if ($client_id === Context::$client_id) {
return static::closeCurrentClient($message);
} // 不是发给当前用户则使用存储中的地址
else {
$address_data = Context::clientIdToAddress($client_id);
if (!$address_data) {
return false;
}
$address = long2ip($address_data['local_ip']) . ":{$address_data['local_port']}";
return static::kickAddress($address, $address_data['connection_id'], $message);
}
}
/**
* 踢掉当前客户端,并以$message通知被踢掉客户端
*
* @param string $message
* @return bool
* @throws Exception
*/
public static function closeCurrentClient($message = null)
{
if (!Context::$connection_id) {
throw new Exception('closeCurrentClient can not be called in async context');
}
$address = long2ip(Context::$local_ip) . ':' . Context::$local_port;
return static::kickAddress($address, Context::$connection_id, $message);
}
/**
* 踢掉某个客户端并直接立即销毁相关连接
*
* @param int $client_id
* @return bool
*/
public static function destoryClient($client_id)
{
if ($client_id === Context::$client_id) {
return static::destoryCurrentClient();
} // 不是发给当前用户则使用存储中的地址
else {
$address_data = Context::clientIdToAddress($client_id);
if (!$address_data) {
return false;
}
$address = long2ip($address_data['local_ip']) . ":{$address_data['local_port']}";
return static::destroyAddress($address, $address_data['connection_id']);
}
}
/**
* 踢掉当前客户端并直接立即销毁相关连接
*
* @return bool
* @throws Exception
*/
public static function destoryCurrentClient()
{
if (!Context::$connection_id) {
throw new Exception('destoryCurrentClient can not be called in async context');
}
$address = long2ip(Context::$local_ip) . ':' . Context::$local_port;
return static::destroyAddress($address, Context::$connection_id);
}
/**
* 将 client_id 与 uid 绑定
*
* @param int $client_id
* @param int|string $uid
* @return void
*/
public static function bindUid($client_id, $uid)
{
static::sendCmdAndMessageToClient($client_id, GatewayProtocol::CMD_BIND_UID, '', $uid);
}
/**
* 将 client_id 与 uid 解除绑定
*
* @param int $client_id
* @param int|string $uid
* @return void
*/
public static function unbindUid($client_id, $uid)
{
static::sendCmdAndMessageToClient($client_id, GatewayProtocol::CMD_UNBIND_UID, '', $uid);
}
/**
* 将 client_id 加入组
*
* @param int $client_id
* @param int|string $group
* @return void
*/
public static function joinGroup($client_id, $group)
{
static::sendCmdAndMessageToClient($client_id, GatewayProtocol::CMD_JOIN_GROUP, '', $group);
}
/**
* 将 client_id 离开组
*
* @param int $client_id
* @param int|string $group
*
* @return void
*/
public static function leaveGroup($client_id, $group)
{
static::sendCmdAndMessageToClient($client_id, GatewayProtocol::CMD_LEAVE_GROUP, '', $group);
}
/**
* 取消分组
*
* @param int|string $group
*
* @return void
*/
public static function ungroup($group)
{
if (!static::isValidGroupId($group)) {
return false;
}
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_UNGROUP;
$gateway_data['ext_data'] = $group;
return static::sendToAllGateway($gateway_data);
}
/**
* 向所有 uid 发送
*
* @param int|string|array $uid
* @param string $message
*
* @return void
*/
public static function sendToUid($uid, $message)
{
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_SEND_TO_UID;
$gateway_data['body'] = $message;
if (!is_array($uid)) {
$uid = array($uid);
}
$gateway_data['ext_data'] = json_encode($uid);
static::sendToAllGateway($gateway_data);
}
/**
* 向 group 发送
*
* @param int|string|array $group 组(不允许是 0 '0' false null array()等为空的值)
* @param string $message 消息
* @param array $exclude_client_id 不给这些client_id发
* @param bool $raw 发送原始数据(即不调用gateway的协议的encode方法)
*
* @return void
*/
public static function sendToGroup($group, $message, $exclude_client_id = null, $raw = false)
{
if (!static::isValidGroupId($group)) {
return false;
}
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_SEND_TO_GROUP;
$gateway_data['body'] = $message;
if ($raw) {
$gateway_data['flag'] |= GatewayProtocol::FLAG_NOT_CALL_ENCODE;
}
if (!is_array($group)) {
$group = array($group);
}
// 分组发送,没有排除的client_id,直接发送
$default_ext_data_buffer = json_encode(array('group' => $group, 'exclude' => null));
if (empty($exclude_client_id)) {
$gateway_data['ext_data'] = $default_ext_data_buffer;
return static::sendToAllGateway($gateway_data);
}
// 分组发送,有排除的client_id,需要将client_id转换成对应gateway进程内的connectionId
if (!is_array($exclude_client_id)) {
$exclude_client_id = array($exclude_client_id);
}
$address_connection_array = static::clientIdArrayToAddressArray($exclude_client_id);
// 如果有businessWorker实例,说明运行在workerman环境中,通过businessWorker中的长连接发送数据
if (static::$businessWorker) {
foreach (static::$businessWorker->gatewayConnections as $address => $gateway_connection) {
$gateway_data['ext_data'] = isset($address_connection_array[$address]) ?
json_encode(array('group' => $group, 'exclude' => $address_connection_array[$address])) :
$default_ext_data_buffer;
/** @var TcpConnection $gateway_connection */
$gateway_connection->send($gateway_data);
}
} // 运行在其它环境中,通过注册中心得到gateway地址
else {
$addresses = static::getAllGatewayAddressesFromRegister();
if (!$addresses) {
throw new Exception('Gateway::getAllGatewayAddressesFromRegister() with registerAddress:' .
static::$registerAddress . ' return ' . var_export($addresses, true));
}
foreach ($addresses as $address) {
$gateway_data['ext_data'] = isset($address_connection_array[$address]) ?
json_encode(array('group' => $group, 'exclude' => $address_connection_array[$address])) :
$default_ext_data_buffer;
static::sendToGateway($address, $gateway_data);
}
}
}
/**
* 更新 session,框架自动调用,开发者不要调用
*
* @param int $client_id
* @param string $session_str
* @return bool
*/
public static function setSocketSession($client_id, $session_str)
{
return static::sendCmdAndMessageToClient($client_id, GatewayProtocol::CMD_SET_SESSION, '', $session_str);
}
/**
* 设置 session,原session值会被覆盖
*
* @param int $client_id
* @param array $session
*
* @return void
*/
public static function setSession($client_id, array $session)
{
if (Context::$client_id === $client_id) {
$_SESSION = $session;
Context::$old_session = $_SESSION;
}
static::setSocketSession($client_id, Context::sessionEncode($session));
}
/**
* 更新 session,实际上是与老的session合并
*
* @param int $client_id
* @param array $session
*
* @return void
*/
public static function updateSession($client_id, array $session)
{
if (Context::$client_id === $client_id) {
$_SESSION = $session + (array)$_SESSION;
Context::$old_session = $_SESSION;
}
static::sendCmdAndMessageToClient($client_id, GatewayProtocol::CMD_UPDATE_SESSION, '', Context::sessionEncode($session));
}
/**
* 获取某个client_id的session
*
* @param int $client_id
* @return mixed false表示出错、null表示用户不存在、array表示具体的session信息
*/
public static function getSession($client_id)
{
$address_data = Context::clientIdToAddress($client_id);
if (!$address_data) {
return false;
}
$address = long2ip($address_data['local_ip']) . ":{$address_data['local_port']}";
if (isset(static::$businessWorker)) {
if (!isset(static::$businessWorker->gatewayConnections[$address])) {
return null;
}
}
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_GET_SESSION_BY_CLIENT_ID;
$gateway_data['connection_id'] = $address_data['connection_id'];
return static::sendAndRecv($address, $gateway_data);
}
/**
* 向某个用户网关发送命令和消息
*
* @param int $client_id
* @param int $cmd
* @param string $message
* @param string $ext_data
* @return boolean
*/
protected static function sendCmdAndMessageToClient($client_id, $cmd, $message, $ext_data = '')
{
// 如果是发给当前用户则直接获取上下文中的地址
if ($client_id === Context::$client_id || $client_id === null) {
$address = long2ip(Context::$local_ip) . ':' . Context::$local_port;
$connection_id = Context::$connection_id;
} else {
$address_data = Context::clientIdToAddress($client_id);
if (!$address_data) {
return false;
}
$address = long2ip($address_data['local_ip']) . ":{$address_data['local_port']}";
$connection_id = $address_data['connection_id'];
}
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = $cmd;
$gateway_data['connection_id'] = $connection_id;
$gateway_data['body'] = $message;
if (!empty($ext_data)) {
$gateway_data['ext_data'] = $ext_data;
}
return static::sendToGateway($address, $gateway_data);
}
/**
* 发送数据并返回
*
* @param int $address
* @param mixed $data
* @return bool
* @throws Exception
*/
protected static function sendAndRecv($address, $data)
{
$buffer = GatewayProtocol::encode($data);
$buffer = static::$secretKey ? static::generateAuthBuffer() . $buffer : $buffer;
$client = stream_socket_client("tcp://$address", $errno, $errmsg, static::$connectTimeout);
if (!$client) {
throw new Exception("can not connect to tcp://$address $errmsg");
}
if (strlen($buffer) === stream_socket_sendto($client, $buffer)) {
$timeout = 5;
// 阻塞读
stream_set_blocking($client, 1);
// 1秒超时
stream_set_timeout($client, 1);
$all_buffer = '';
$time_start = microtime(true);
$pack_len = 0;
while (1) {
$buf = stream_socket_recvfrom($client, 655350);
if ($buf !== '' && $buf !== false) {
$all_buffer .= $buf;
} else {
if (feof($client)) {
throw new Exception("connection close tcp://$address");
} elseif (microtime(true) - $time_start > $timeout) {
break;
}
continue;
}
$recv_len = strlen($all_buffer);
if (!$pack_len && $recv_len >= 4) {
$pack_len = current(unpack('N', $all_buffer));
}
// 回复的数据都是以\n结尾
if (($pack_len && $recv_len >= $pack_len + 4) || microtime(true) - $time_start > $timeout) {
break;
}
}
// 返回结果
return unserialize(substr($all_buffer, 4));
} else {
throw new Exception("sendAndRecv($address, \$bufer) fail ! Can not send data!", 502);
}
}
/**
* 发送数据到网关
*
* @param string $address
* @param array $gateway_data
* @return bool
*/
protected static function sendToGateway($address, $gateway_data)
{
return static::sendBufferToGateway($address, GatewayProtocol::encode($gateway_data));
}
/**
* 发送buffer数据到网关
* @param string $address
* @param string $gateway_buffer
* @return bool
*/
protected static function sendBufferToGateway($address, $gateway_buffer)
{
// 有$businessWorker说明是workerman环境,使用$businessWorker发送数据
if (static::$businessWorker) {
if (!isset(static::$businessWorker->gatewayConnections[$address])) {
return false;
}
return static::$businessWorker->gatewayConnections[$address]->send($gateway_buffer, true);
}
// 非workerman环境
$gateway_buffer = static::$secretKey ? static::generateAuthBuffer() . $gateway_buffer : $gateway_buffer;
$flag = static::$persistentConnection ? STREAM_CLIENT_PERSISTENT | STREAM_CLIENT_CONNECT : STREAM_CLIENT_CONNECT;
$client = stream_socket_client("tcp://$address", $errno, $errmsg, static::$connectTimeout, $flag);
return strlen($gateway_buffer) == stream_socket_sendto($client, $gateway_buffer);
}
/**
* 向所有 gateway 发送数据
*
* @param string $gateway_data
* @throws Exception
*
* @return void
*/
protected static function sendToAllGateway($gateway_data)
{
$buffer = GatewayProtocol::encode($gateway_data);
// 如果有businessWorker实例,说明运行在workerman环境中,通过businessWorker中的长连接发送数据
if (static::$businessWorker) {
foreach (static::$businessWorker->gatewayConnections as $gateway_connection) {
/** @var TcpConnection $gateway_connection */
$gateway_connection->send($buffer, true);
}
} // 运行在其它环境中,通过注册中心得到gateway地址
else {
$all_addresses = static::getAllGatewayAddressesFromRegister();
if (!$all_addresses) {
throw new Exception('Gateway::getAllGatewayAddressesFromRegister() with registerAddress:' .
static::$registerAddress . ' return ' . var_export($all_addresses, true));
}
foreach ($all_addresses as $address) {
static::sendBufferToGateway($address, $buffer);
}
}
}
/**
* 踢掉某个网关的 socket
*
* @param string $address
* @param int $connection_id
* @return bool
*/
protected static function kickAddress($address, $connection_id, $message)
{
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_KICK;
$gateway_data['connection_id'] = $connection_id;
$gateway_data['body'] = $message;
return static::sendToGateway($address, $gateway_data);
}
/**
* 销毁某个网关的 socket
*
* @param string $address
* @param int $connection_id
* @return bool
*/
protected static function destroyAddress($address, $connection_id)
{
$gateway_data = GatewayProtocol::$empty;
$gateway_data['cmd'] = GatewayProtocol::CMD_DESTROY;
$gateway_data['connection_id'] = $connection_id;
return static::sendToGateway($address, $gateway_data);
}
/**
* 将clientid数组转换成address数组
*
* @param array $client_id_array
* @return array
*/
protected static function clientIdArrayToAddressArray(array $client_id_array)
{
$address_connection_array = array();
foreach ($client_id_array as $client_id) {
$address_data = Context::clientIdToAddress($client_id);
if ($address_data) {
$address = long2ip($address_data['local_ip']) .
":{$address_data['local_port']}";
$address_connection_array[$address][$address_data['connection_id']] = $address_data['connection_id'];
}
}
return $address_connection_array;
}
/**
* 设置 gateway 实例
*
* @param \GatewayWorker\BusinessWorker $business_worker_instance
*/
public static function setBusinessWorker($business_worker_instance)
{
static::$businessWorker = $business_worker_instance;
}
/**
* 获取通过注册中心获取所有 gateway 通讯地址
*
* @return array
* @throws Exception
*/
protected static function getAllGatewayAddressesFromRegister()
{
static $addresses_cache, $last_update, $last_register_address;
$time_now = time();
$expiration_time = 1;
if (empty($addresses_cache) || $time_now - $last_update > $expiration_time || $last_register_address !== static::$registerAddress) {
$client = stream_socket_client('tcp://' . static::$registerAddress, $errno, $errmsg, static::$connectTimeout);
if (!$client) {
throw new Exception('Can not connect to tcp://' . static::$registerAddress . ' ' . $errmsg);
}
fwrite($client, '{"event":"worker_connect","secret_key":"' . static::$secretKey . '"}' . "\n");
stream_set_timeout($client, 5);
$ret = fgets($client, 655350);
if (!$ret || !$data = json_decode(trim($ret), true)) {
throw new Exception('getAllGatewayAddressesFromRegister fail. tcp://' .
static::$registerAddress . ' return ' . var_export($ret, true));
}
$last_update = $time_now;
$last_register_address = static::$registerAddress;
$addresses_cache = $data['addresses'];
}
return $addresses_cache;
}
/**
* 检查群组id是否合法
*
* @param $group
* @return bool
*/
protected static function isValidGroupId($group)
{
if (empty($group)) {
echo new \Exception('group(' . var_export($group, true) . ') empty');
return false;
}
return true;
}
}
/**
* 上下文 包含当前用户uid, 内部通信local_ip local_port socket_id ,以及客户端client_ip client_port
*/
class Context
{
/**
* 内部通讯id
* @var string
*/
public static $local_ip;
/**
* 内部通讯端口
* @var int
*/
public static $local_port;
/**
* 客户端ip
* @var string
*/
public static $client_ip;
/**
* 客户端端口
* @var int
*/
public static $client_port;
/**
* client_id
* @var string
*/
public static $client_id;
/**
* 连接connection->id
* @var int
*/
public static $connection_id;
/**
* 旧的session
*
* @var string
*/
public static $old_session;
/**
* 编码session
* @param mixed $session_data
* @return string
*/
public static function sessionEncode($session_data = '')
{
if ($session_data !== '') {
return serialize($session_data);
}
return '';
}
/**
* 解码session
* @param string $session_buffer
* @return mixed
*/
public static function sessionDecode($session_buffer)
{
return unserialize($session_buffer);
}
/**
* 清除上下文
* @return void
*/
public static function clear()
{
static::$local_ip = static::$local_port = static::$client_ip = static::$client_port =
static::$client_id = static::$connection_id = static::$old_session = null;
}
/**
* 通讯地址到client_id的转换
* @return string
*/
public static function addressToClientId($local_ip, $local_port, $connection_id)
{
return bin2hex(pack('NnN', $local_ip, $local_port, $connection_id));
}
/**
* client_id到通讯地址的转换
* @return array
*/
public static function clientIdToAddress($client_id)
{
if (strlen($client_id) !== 20) {
throw new \Exception("client_id $client_id is invalid");
}
return unpack('Nlocal_ip/nlocal_port/Nconnection_id', pack('H*', $client_id));
}
}
/**
* Gateway 与 Worker 间通讯的二进制协议
*
* struct GatewayProtocol
* {
* unsigned int pack_len,
* unsigned char cmd,//命令字
* unsigned int local_ip,
* unsigned short local_port,
* unsigned int client_ip,
* unsigned short client_port,
* unsigned int connection_id,
* unsigned char flag,
* unsigned short gateway_port,
* unsigned int ext_len,
* char[ext_len] ext_data,
* char[pack_length-HEAD_LEN] body//包体
* }
* NCNnNnNCnN
*/
class GatewayProtocol
{
// 发给worker,gateway有一个新的连接
const CMD_ON_CONNECT = 1;
// 发给worker的,客户端有消息
const CMD_ON_MESSAGE = 3;
// 发给worker上的关闭链接事件
const CMD_ON_CLOSE = 4;
// 发给gateway的向单个用户发送数据
const CMD_SEND_TO_ONE = 5;
// 发给gateway的向所有用户发送数据
const CMD_SEND_TO_ALL = 6;
// 发给gateway的踢出用户
// 1、如果有待发消息,将在发送完后立即销毁用户连接
// 2、如果无待发消息,将立即销毁用户连接
const CMD_KICK = 7;
// 发给gateway的立即销毁用户连接
const CMD_DESTROY = 8;
// 发给gateway,通知用户session更新
const CMD_UPDATE_SESSION = 9;
// 获取在线状态
const CMD_GET_ALL_CLIENT_SESSIONS = 10;
// 判断是否在线
const CMD_IS_ONLINE = 11;
// client_id绑定到uid
const CMD_BIND_UID = 12;
// 解绑
const CMD_UNBIND_UID = 13;
// 向uid发送数据
const CMD_SEND_TO_UID = 14;
// 根据uid获取绑定的clientid
const CMD_GET_CLIENT_ID_BY_UID = 15;
// 加入组
const CMD_JOIN_GROUP = 20;
// 离开组
const CMD_LEAVE_GROUP = 21;
// 向组成员发消息
const CMD_SEND_TO_GROUP = 22;
// 获取组成员
const CMD_GET_CLINET_SESSUONS_BY_GROUP = 23;
// 获取组在线连接数
const CMD_GET_CLIENT_COUNT_BY_GROUP = 24;
// 按照条件查找
const CMD_SELECT = 25;
// 获取在线的群组ID
const CMD_GET_GROUP_ID_LIST = 26;
// 取消分组
const CMD_UNGROUP = 27;
// worker连接gateway事件
const CMD_WORKER_CONNECT = 200;
// 心跳
const CMD_PING = 201;
// GatewayClient连接gateway事件
const CMD_GATEWAY_CLIENT_CONNECT = 202;
// 根据client_id获取session
const CMD_GET_SESSION_BY_CLIENT_ID = 203;
// 发给gateway,覆盖session
const CMD_SET_SESSION = 204;
// 当websocket握手时触发,只有websocket协议支持此命令字
const CMD_ON_WEBSOCKET_CONNECT = 205;
// 包体是标量
const FLAG_BODY_IS_SCALAR = 0x01;
// 通知gateway在send时不调用协议encode方法,在广播组播时提升性能
const FLAG_NOT_CALL_ENCODE = 0x02;
/**
* 包头长度
*
* @var int
*/
const HEAD_LEN = 28;
public static $empty = array(
'cmd' => 0,
'local_ip' => 0,
'local_port' => 0,
'client_ip' => 0,
'client_port' => 0,
'connection_id' => 0,
'flag' => 0,
'gateway_port' => 0,
'ext_data' => '',
'body' => '',
);
/**
* 返回包长度
*
* @param string $buffer
* @return int return current package length
*/
public static function input($buffer)
{
if (strlen($buffer) < self::HEAD_LEN) {
return 0;
}
$data = unpack("Npack_len", $buffer);
return $data['pack_len'];
}
/**
* 获取整个包的 buffer
*
* @param mixed $data
* @return string
*/
public static function encode($data)
{
$flag = (int)is_scalar($data['body']);
if (!$flag) {
$data['body'] = serialize($data['body']);
}
$data['flag'] |= $flag;
$ext_len = strlen($data['ext_data']);
$package_len = self::HEAD_LEN + $ext_len + strlen($data['body']);
return pack("NCNnNnNCnN", $package_len,
$data['cmd'], $data['local_ip'],
$data['local_port'], $data['client_ip'],
$data['client_port'], $data['connection_id'],
$data['flag'], $data['gateway_port'],
$ext_len) . $data['ext_data'] . $data['body'];
}
/**
* 从二进制数据转换为数组
*
* @param string $buffer
* @return array
*/
public static function decode($buffer)
{
$data = unpack("Npack_len/Ccmd/Nlocal_ip/nlocal_port/Nclient_ip/nclient_port/Nconnection_id/Cflag/ngateway_port/Next_len",
$buffer);
if ($data['ext_len'] > 0) {
$data['ext_data'] = substr($buffer, self::HEAD_LEN, $data['ext_len']);
if ($data['flag'] & self::FLAG_BODY_IS_SCALAR) {
$data['body'] = substr($buffer, self::HEAD_LEN + $data['ext_len']);
} else {
$data['body'] = unserialize(substr($buffer, self::HEAD_LEN + $data['ext_len']));
}
} else {
$data['ext_data'] = '';
if ($data['flag'] & self::FLAG_BODY_IS_SCALAR) {
$data['body'] = substr($buffer, self::HEAD_LEN);
} else {
$data['body'] = unserialize(substr($buffer, self::HEAD_LEN));
}
}
return $data;
}
}
...@@ -16,6 +16,8 @@ class ConfirmOpenDoor implements Command ...@@ -16,6 +16,8 @@ class ConfirmOpenDoor implements Command
private $orderId = ''; private $orderId = '';
private $operationType = 1;
public function getMessage(): string public function getMessage(): string
{ {
...@@ -24,6 +26,8 @@ class ConfirmOpenDoor implements Command ...@@ -24,6 +26,8 @@ class ConfirmOpenDoor implements Command
'command' => $this->command, 'command' => $this->command,
'order_id' => $this->orderId, 'order_id' => $this->orderId,
'hdg_topic' => 'open_door',
'operation_type' => $this->operationType,
]); ]);
} }
...@@ -60,5 +64,21 @@ class ConfirmOpenDoor implements Command ...@@ -60,5 +64,21 @@ class ConfirmOpenDoor implements Command
$this->orderId = $orderId; $this->orderId = $orderId;
} }
/**
* @return int
*/
public function getOperationType(): int
{
return $this->operationType;
}
/**
* @param int $operationType
*/
public function setOperationType(int $operationType): void
{
$this->operationType = $operationType;
}
} }
\ No newline at end of file
...@@ -9,10 +9,23 @@ ...@@ -9,10 +9,23 @@
namespace Lib\command; namespace Lib\command;
use Src\box\Box;
use Src\box\door\Door;
class OpenDoor implements Command class OpenDoor implements Command
{ {
private $command = '11'; private $command = '11';
/**
* @var Box
*/
private $box;
/**
* @var Door
*/
private $door;
private $boxNum = 0; private $boxNum = 0;
private $doorNum = 0; private $doorNum = 0;
...@@ -21,23 +34,70 @@ class OpenDoor implements Command ...@@ -21,23 +34,70 @@ class OpenDoor implements Command
private $operation = 1; private $operation = 1;
private $operationType = 1;
private $boxId = 0;
private $doorId = 0;
public function getMessage(): string public function getMessage(): string
{ {
return json_encode([ return json_encode([
'command' => $this->command, 'command' => $this->command,
'hdg_topic' => 'open_door',
'doors' => [ 'doors' => [
'box_num' => $this->boxNum, 'box_num' => $this->boxNum,
'door_num' => $this->doorNum, 'door_num' => $this->doorNum,
'order_id' => $this->orderId, 'order_id' => $this->orderId,
'operation' => $this->operation 'operation' => $this->operation,
'door_id' => $this->doorId,
'box_id' => $this->boxId,
'operation_type' => $this->operationType,
] ]
]); ]);
} }
/** /**
* @return Box
*/
public function getBox(): Box
{
return $this->box;
}
/**
* @param Box $box
*/
public function setBox(Box $box): void
{
$this->boxId = $box->getId();
$this->boxNum = $box->getName();
$this->box = $box;
}
/**
* @return Door
*/
public function getDoor(): Door
{
return $this->door;
}
/**
* @param Door $door
*/
public function setDoor(Door $door): void
{
$this->doorId = $door->getId();
$this->doorNum = $door->getName();
$this->door = $door;
}
/**
* @return string * @return string
*/ */
public function getCommand(): string public function getCommand(): string
...@@ -117,5 +177,53 @@ class OpenDoor implements Command ...@@ -117,5 +177,53 @@ class OpenDoor implements Command
$this->operation = $operation; $this->operation = $operation;
} }
/**
* @return int
*/
public function getOperationType(): int
{
return $this->operationType;
}
/**
* @param int $operationType
*/
public function setOperationType(int $operationType): void
{
$this->operationType = $operationType;
}
/**
* @return int
*/
public function getBoxId(): int
{
return $this->boxId;
}
/**
* @param int $boxId
*/
public function setBoxId(int $boxId): void
{
$this->boxId = $boxId;
}
/**
* @return int
*/
public function getDoorId(): int
{
return $this->doorId;
}
/**
* @param int $doorId
*/
public function setDoorId(int $doorId): void
{
$this->doorId = $doorId;
}
} }
\ No newline at end of file
...@@ -9,6 +9,7 @@ ...@@ -9,6 +9,7 @@
namespace Lib\messageId; namespace Lib\messageId;
use Illuminate\Support\Facades\Log;
use Mockery\Exception; use Mockery\Exception;
class MessageExist class MessageExist
...@@ -65,7 +66,7 @@ class MessageExist ...@@ -65,7 +66,7 @@ class MessageExist
{ {
$redis = app('Redis'); $redis = app('Redis');
if ($redis instanceof \Redis) { if ($redis instanceof \Redis) {
$redis->zAdd($this->todayKey(), $this->state, $this->id); $redis->zAdd($this->todayKey(), (int)$state, $id);
if ($redis->exists($this->yesterdayKey())) { if ($redis->exists($this->yesterdayKey())) {
$redis->setTimeout($this->yesterdayKey(), 7 * 60 * 60 * 24); $redis->setTimeout($this->yesterdayKey(), 7 * 60 * 60 * 24);
} }
......
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/10/11
* Time: 下午3:05
*/
namespace Lib\web_socket;
abstract class Message
{
protected $command = '';
abstract public function getMessage(): string;
}
\ No newline at end of file
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/10/11
* Time: 下午3:06
*/
namespace Lib\web_socket;
class OpenDoorResult extends Message
{
private $startTime = '';
/**
* @var string
*/
private $outTradeNo = '';
/**
* @var string
*/
private $operation = 'open';
/**
* @var string
*/
private $result = 'FAIL';
private $takeBoxNum = 0;
private $takeDoorNum = 0;
/**
* @return string
*/
public function getMessage(): string
{
return json_encode([
'command' => $this->command,
'data' => [
'out_trade_no' => $this->outTradeNo,
'operation' => $this->operation,
'result' => $this->result,
'start_time' => $this->startTime,
'take_box_num' => $this->takeBoxNum,
'take_door_num' => $this->takeDoorNum,
],
]);
}
/**
* @return int
*/
public function getTakeBoxNum(): int
{
return $this->takeBoxNum;
}
/**
* @param int $takeBoxNum
*/
public function setTakeBoxNum(int $takeBoxNum): void
{
$this->takeBoxNum = $takeBoxNum;
}
/**
* @return int
*/
public function getTakeDoorNum(): int
{
return $this->takeDoorNum;
}
/**
* @param int $takeDoorNum
*/
public function setTakeDoorNum(int $takeDoorNum): void
{
$this->takeDoorNum = $takeDoorNum;
}
/**
* @return string
*/
public function getOutTradeNo(): string
{
return $this->outTradeNo;
}
/**
* @param string $outTradeNo
*/
public function setOutTradeNo(string $outTradeNo): void
{
$this->outTradeNo = $outTradeNo;
}
/**
* @return string
*/
public function getOperation(): string
{
return $this->operation;
}
/**
* @param string $operation
*/
public function setOperation(string $operation): void
{
$this->operation = $operation;
}
/**
* @return string
*/
public function getResult(): string
{
return $this->result;
}
/**
* @param string $result
*/
public function setResult(string $result): void
{
$this->result = $result;
}
}
\ No newline at end of file
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/10/11
* Time: 下午3:05
*/
namespace Lib\web_socket;
class PushClient
{
}
\ No newline at end of file
...@@ -16,20 +16,24 @@ $router->get('/', function () use ($router) { ...@@ -16,20 +16,24 @@ $router->get('/', function () use ($router) {
return $router->app->version(); return $router->app->version();
}); });
$router->group(['middleware' => ['mqtt']], function () use ($router) { $router->group(['middleware' => ['mqtt', 'message_exist']], function () use ($router) {
$router->get('ac/version_info', 'ac\VersionController@get'); $router->get('ac/version_info', 'ac\VersionController@get');
$router->post('ac/version_info', 'ac\VersionController@get'); $router->post('ac/version_info', 'ac\VersionController@get');
$router->get('ac/state', 'ac\StateController@get'); $router->get('ac/state', 'ac\StateController@get');
$router->post('ac/state', 'ac\StateController@post'); $router->post('ac/state', 'ac\StateController@post');
$router->post('door/lock/version', 'door\LockController@version'); $router->post('door/lock/version', 'door\LockController@version');
$router->post('door/state', 'door\StateController@post'); $router->post('door/state', 'door\StateController@state');
$router->post('ac/alarm/info', 'ac\AlarmController@post'); $router->post('ac/alarm/info', 'ac\AlarmController@post');
$router->post('update', 'UpdateController@update'); $router->post('update', 'UpdateController@update');
$router->post('battery/update', 'BatteryController@update');
}); });
$router->get('test', 'TestController@test');
$router->post('door/opening', 'door\OpenController@post'); $router->post('door/opening', 'door\OpenController@post');
$router->post('box/init', 'door\InitController@init'); $router->post('box/init', 'door\InitController@init');
......
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/10/8
* Time: 上午11:11
*/
namespace Src;
class Site
{
private $id = 0;
private $name = '';
/**
* @return int
*/
public function getId(): int
{
return $this->id;
}
/**
* @param int $id
*/
public function setId(int $id): void
{
$this->id = $id;
}
/**
* @return string
*/
public function getName(): string
{
return $this->name;
}
/**
* @param string $name
*/
public function setName(string $name): void
{
$this->name = $name;
}
}
\ No newline at end of file
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/10/8
* Time: 上午11:12
*/
namespace Src\battery;
class Battery
{
use SearchMaxQuantity;
private $id = 0;
private $code = '';
private $voltage = 0;
private $quantity = 0;
private $mosTmp = 0;
private $maxTmp = 0;
private $minTmp = 0;
private $reserve = 0;
private $boxNum = 0;
private $doorNum = 0;
public static function getRedisKey(string $boxCode)
{
return config('redis_key.BATTERY_QUANTITY') . ':' . $boxCode;
}
public function updateQuantity(int $quantity)
{
$redis = app('Redis');
if ($redis instanceof \Redis) {
$redis->zAdd(static::getRedisKey($this->boxCode), $quantity, $this->id);
}
}
/**
* @return string
*/
public function getCode(): string
{
return $this->code;
}
/**
* @param string $code
*/
public function setCode(string $code): void
{
$this->code = $code;
}
/**
* @return int
*/
public function getVoltage(): int
{
return $this->voltage;
}
/**
* @param int $voltage
*/
public function setVoltage(int $voltage): void
{
$this->voltage = $voltage;
}
/**
* @return int
*/
public function getQuantity(): int
{
return $this->quantity;
}
/**
* @param int $quantity
*/
public function setQuantity(int $quantity): void
{
$this->quantity = $quantity;
}
/**
* @return int
*/
public function getMosTmp(): int
{
return $this->mosTmp;
}
/**
* @param int $mosTmp
*/
public function setMosTmp(int $mosTmp): void
{
$this->mosTmp = $mosTmp;
}
/**
* @return int
*/
public function getMaxTmp(): int
{
return $this->maxTmp;
}
/**
* @param int $maxTmp
*/
public function setMaxTmp(int $maxTmp): void
{
$this->maxTmp = $maxTmp;
}
/**
* @return int
*/
public function getMinTmp(): int
{
return $this->minTmp;
}
/**
* @param int $minTmp
*/
public function setMinTmp(int $minTmp): void
{
$this->minTmp = $minTmp;
}
/**
* @return int
*/
public function getReserve(): int
{
return $this->reserve;
}
/**
* @param int $reserve
*/
public function setReserve(int $reserve): void
{
$this->reserve = $reserve;
}
/**
* @return int
*/
public function getBoxNum(): int
{
return $this->boxNum;
}
/**
* @param int $boxNum
*/
public function setBoxNum(int $boxNum): void
{
$this->boxNum = $boxNum;
}
/**
* @return int
*/
public function getDoorNum(): int
{
return $this->doorNum;
}
/**
* @param int $doorNum
*/
public function setDoorNum(int $doorNum): void
{
$this->doorNum = $doorNum;
}
/**
* @return int
*/
public function getId(): int
{
return $this->id;
}
/**
* @param int $id
*/
public function setId(int $id): void
{
$this->id = $id;
}
}
\ No newline at end of file
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/10/12
* Time: 上午12:49
*/
namespace src\battery;
trait SearchMaxQuantity
{
private $boxCode = '';
/**
* 查找电柜中电量最足的电池
* @param string $boxCode
* @return int
*/
public function searchMaxQuantity(string $boxCode): int
{
$this->boxCode = $boxCode;
$redis = app('Redis');
if ($redis instanceof \Redis) {
$max = $redis->zRange(Battery::getRedisKey($boxCode), -1, -1);
return blank($max) ? 0 : current($max);
}
}
/**
* @param string $batteryCode
* @param string $boxCode
* @return int
*/
public function getBatteryQuantity(string $batteryCode, string $boxCode): int
{
$redis = app('Redis');
if ($redis instanceof \Redis) {
$quantity = $redis->zScore(Battery::getRedisKey($boxCode), $batteryCode);
return blank($quantity) ? 0 : $quantity;
}
}
}
\ No newline at end of file
...@@ -37,9 +37,14 @@ class Box ...@@ -37,9 +37,14 @@ class Box
} }
public function __get($name)
{
return $this->$name;
}
public static function getOpenDoorTopic(string $boxCode) public static function getOpenDoorTopic(string $boxCode)
{ {
return env('MQTT_PRODUCT_KEY') . '/' . $boxCode . '/open_door'; return env('MQTT_PRODUCT_KEY') . '/' . $boxCode . '/hdg_get';
} }
......
...@@ -14,6 +14,11 @@ use Log; ...@@ -14,6 +14,11 @@ use Log;
class Door class Door
{ {
const OPERATION = [
'close' => 0,
'open' => 1,
];
const STATE = [ const STATE = [
...@@ -55,6 +60,11 @@ class Door ...@@ -55,6 +60,11 @@ class Door
private $boxNum = 0; private $boxNum = 0;
public function __get($name)
{
return $this->$name;
}
/** /**
* @var int * @var int
* 1: 开门 * 1: 开门
......
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/10/16
* Time: 上午9:36
*/
namespace src\client;
use Illuminate\Http\Request;
interface Client
{
public function handle(Request $request);
}
\ No newline at end of file
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/10/16
* Time: 上午9:56
*/
namespace src\client\open_door;
use Illuminate\Http\Request;
use Illuminate\Support\Facades\Log;
use Lib\aliyun\api\PublishToTopic;
use Lib\command\OpenDoor;
use Src\battery\Battery;
use Src\box\Box;
use Src\box\door\Door;
use src\client\Client;
use Src\order\Order;
use Src\order\OrderCache;
use Src\pay_rule\CashPay;
class ReturnBattery implements Client
{
public function handle(Request $request)
{
$payload = $request->get('payload');
$payload = $payload['doors'];
$outTradeNo = $payload['order_id'];
$batteryCode = $payload['battery_code'];
$doorNum = $payload['door_num'];
$boxNum = $payload['box_num'];
$boxCode = $request->get('box_code');
$operationType = $payload['operation_type'];
$orderCache = new OrderCache();
$order = new Order();
$order->setOutTradeNo($outTradeNo);
$orderCache = new OrderCache();
$orderCache->setOrder($order);
$cache = $orderCache->getCache();
if (blank($cache) === true) {
Log::error('can not find order cache ---' . $outTradeNo);
return;
}
switch ($operationType) {
case 1:
if (blank($batteryCode) === true) {
//todo 提示未归还电池
Log::info('close return door without battery');
return;
} else {
$returnBoxId = $cache['returnBoxId'];
$returnDoorId = $cache['returnDoorId'];
//todo 更新订单结算支付金额,存入数据库
$orderModel = OrderModel::where('code', $outTradeNo)->firt();
$cashRule = new CashPay();
$cashRule->setOrderModel($orderModel);
$payAmount = $cashRule->getPayAmount();
$date = date('Y-m-d H:i:s');
DB::table('orders')->where('code', $outTradeNo)->update([
'in_date' => $date,
'in_box_id' => $returnBoxId,
'in_room_id' => $returnDoorId,
'in_battery_code' => $batteryCode,
'rent_money' => $payAmount,
'use_money' => $payAmount,
'in_room_name' => $boxNum,
'in_site_name' => '利尔达', //fixme 从boxcode反查获得,
'in_site_id' => 1, //fixme
]);
//todo web socket 推送还电成功,打开新的电池柜舱门
//todo 更新电池归属,更新电池电量
//todo 查找电量最大的电池仓下发打开命令
$orderCache = new OrderCache();
$order->setOutTradeNo($outTradeNo);
$cache = $orderCache->getCache();
$userId = $cache['useId'];
$battery = new Battery();
$maxBatteryId = $battery->searchMaxQuantity($boxCode);
$batteryModel = BatteryModel::find($maxBatteryId);
$outTradeNo = $boxCode . rand(10000, 999999);
$takeBox = new Box();
$takeBox->setName($batteryModel->box_type);
$takeBox->setId($batteryModel->box_id);
$takeDoor = new Door();
$takeDoor->setName($batteryModel->door_name);
$takeDoor->setId($batteryModel->room_id);
$order->setOutTradeNo($outTradeNo);
$order->setTakeBox($takeBox);
$order->setTakeDoor($takeDoor);
$order->setUserId($userId);
$orderModel = new OrderModel();
$orderModel->initOrder($order);
$openDoorCommand = new OpenDoor();
$openDoorCommand->setOrderId($order->getOutTradeNo());
$openDoorCommand->setBoxNum($order->getTakeBox()->getName());
$openDoorCommand->setDoorNum($order->getTakeDoor()->getName());
$openDoorCommand->setDoorId($order->getTakeDoor()->getId());
$openDoorCommand->setBoxId($order->getTakeBox()->getId());
$publishToTopic = new PublishToTopic();
$publishToTopic->setMessageContent($openDoorCommand->getMessage());
$publishToTopic->setProductKey(env('MQTT_PRODUCT_KEY'));
$publishToTopic->setTopicFullName(Box::getOpenDoorTopic($boxCode));
//todo websocket推送消息 取出舱门打开
}
}
}
}
\ No newline at end of file
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/10/16
* Time: 上午9:35
*/
namespace src\client\open_door;
use Illuminate\Http\Request;
use Lib\aliyun\api\PublishToTopic;
use Lib\command\ConfirmOpenDoor;
use Src\box\Box;
use src\client\Client;
use Src\order\OrderCache;
/**
* 确认归还电池舱门已经打开
* Class ReturnOpenDoor
* @package src\client\open_door
*/
class ReturnOpenDoor implements Client
{
public function handle(Request $request)
{
$payload = $request->get('payload');
$door = $payload['doors'];
$boxNum = $door['box_num'];
$doorNum = $door['door_num'];
$operation = $door['operation'];
$operationType = $door['operation_type'];
$orderId = $door['order_id'];
$boxCode = $request->get('box_code');
$boxId = $request->get('box_id');
$doorId = $request->get('door_id');
if ($operation !== Door::OPERATION['open']) {
//todo 根据operationType 发送打开舱门失败web socket.回复确认消息
} else {
//todo 根据operationType打开舱门成功推送web socket
}
switch ($operationType) {
case 1:
$orderCache = new OrderCache();
$orderCache->update([
'returnBoxId' => $boxId,
'returnDoorId' => $doorId,
'returnBoxNum' => $boxNum,
'returnDoorNum' => $doorNum
]);
//todo 检查订单是否存在,websocket推送打开门成功,检查电池是否放入,更新订单状态
$confirmOpenDoor = new ConfirmOpenDoor();
$confirmOpenDoor->setOrderId($orderId);
$publishToTopic = new PublishToTopic();
$publishToTopic->setMessageContent($confirmOpenDoor->getMessage());
$publishToTopic->setProductKey(env('MQTT_PRODUCT_KEY'));
$publishToTopic->setTopicFullName(Box::getOpenDoorTopic($boxCode));
$client = new \Lib\aliyun\Client();
$client->setApi($publishToTopic);
if ($client->request() === true) {
Log::debug('confirm order -' . $orderId);
}
break;
case 0:
break;
default:
break;
}
}
}
\ No newline at end of file
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/10/16
* Time: 上午9:56
*/
namespace src\client\open_door;
class TakeOpenDoor
{
}
\ No newline at end of file
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/10/7
* Time: 下午9:23
*/
namespace Src\order;
use Src\battery\Battery;
use Src\box\Box;
use Src\box\door\Door;
use Src\Site;
class Order
{
/**
* @var int
*/
private $id = 0;
/**
* @var string
*/
private $outTradeNo = '';
/**
* @var int
*/
private $userId = 0;
/**
* @var Site
*/
private $site;
/**
* @var string
*/
private $boxCode = '';
/**
* @var Box
*/
private $openBox;
/**
* @var Box
*/
private $returnBox;
/**
* @var Door
*/
private $openDoor;
/**
* @var Door
*/
private $returnDoor;
/**
* @var Battery
*/
private $returnBattery;
/**
* @var Door
*/
private $takeDoor;
/**
* @var Box
*/
private $takeBox;
/**
* @var Battery
*/
private $takeBattery;
public function __get($name)
{
return $this->$name;
}
/**
* @return int
*/
public function getId(): int
{
return $this->id;
}
/**
* @param int $id
*/
public function setId(int $id): void
{
$this->id = $id;
}
/**
* @return string
*/
public function getOutTradeNo(): string
{
return $this->outTradeNo;
}
/**
* @param string $outTradeNo
*/
public function setOutTradeNo(string $outTradeNo): void
{
$this->outTradeNo = $outTradeNo;
}
/**
* @return int
*/
public function getUserId(): int
{
return $this->userId;
}
/**
* @param int $userId
*/
public function setUserId(int $userId): void
{
$this->userId = $userId;
}
/**
* @return Site
*/
public function getSite(): Site
{
return $this->site;
}
/**
* @param Site $site
*/
public function setSite(Site $site): void
{
$this->site = $site;
}
/**
* @return string
*/
public function getBoxCode(): string
{
return $this->boxCode;
}
/**
* @param string $boxCode
*/
public function setBoxCode(string $boxCode): void
{
$this->boxCode = $boxCode;
}
/**
* @return Box
*/
public function getOpenBox(): Box
{
return $this->openBox;
}
/**
* @param Box $openBox
*/
public function setOpenBox(Box $openBox): void
{
$this->openBox = $openBox;
}
/**
* @return Box
*/
public function getReturnBox(): Box
{
return $this->returnBox;
}
/**
* @param Box $returnBox
*/
public function setReturnBox(Box $returnBox): void
{
$this->returnBox = $returnBox;
}
/**
* @return Door
*/
public function getOpenDoor(): Door
{
return $this->openDoor;
}
/**
* @param Door $openDoor
*/
public function setOpenDoor(Door $openDoor): void
{
$this->openDoor = $openDoor;
}
/**
* @return Door
*/
public function getReturnDoor(): Door
{
return $this->returnDoor;
}
/**
* @param Door $returnDoor
*/
public function setReturnDoor(Door $returnDoor): void
{
$this->returnDoor = $returnDoor;
}
/**
* @return Battery
*/
public function getReturnBattery(): Battery
{
return $this->returnBattery;
}
/**
* @param Battery $returnBattery
*/
public function setReturnBattery(Battery $returnBattery): void
{
$this->returnBattery = $returnBattery;
}
/**
* @return Battery
*/
public function getTakeBattery(): Battery
{
return $this->takeBattery;
}
/**
* @param Battery $takeBattery
*/
public function setTakeBattery(Battery $takeBattery): void
{
$this->takeBattery = $takeBattery;
}
/**
* @return Door
*/
public function getTakeDoor(): Door
{
return $this->takeDoor;
}
/**
* @param Door $takeDoor
*/
public function setTakeDoor(Door $takeDoor): void
{
$this->takeDoor = $takeDoor;
}
/**
* @return Box
*/
public function getTakeBox(): Box
{
return $this->takeBox;
}
/**
* @param Box $takeBox
*/
public function setTakeBox(Box $takeBox): void
{
$this->takeBox = $takeBox;
}
}
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/10/11
* Time: 下午4:05
*/
namespace Src\order;
use Src\order\Order;
class OrderCache
{
/**
* @var Order
*/
private $order;
private $startOpenReturnDoorTime = '';
private $startOpenTakeDoorTime = '';
private $state;
/**
* @var \Redis
*/
private $redis;
public function __construct()
{
$this->redis = app('Redis');
}
public function getRedisKey()
{
return config('redis_key.ORDER_CACHE') . ':' . $this->order->getOutTradeNo();
}
public function update(array $attribute)
{
$this->redis->hMset($this->getRedisKey(), $attribute);
return $this;
}
/**
* @return array
*/
public function getCache()
{
return $this->redis->hGetAll($this->getRedisKey());
}
public function saveState()
{
$this->redis->hSet($this->getRedisKey(), 'state', $this->state);
return $this;
}
/**
* @return $this
*/
public function saveStartOpenReturnDoorTime()
{
$this->redis->hSet($this->getRedisKey(), 'startOpenReturnDoorTime', $this->startOpenReturnDoorTime);
return $this;
}
public function saveOrder($attribute = '')
{
if (is_string($attribute)) {
$this->redis->hSet($this->getRedisKey(), $attribute, $this->order->$attribute);
} elseif (is_array($attribute)) {
foreach ($attribute as $item) {
$this->redis->hSet($this->getRedisKey(), $item, $this->order->$item);
}
}
}
public function getOrderCache()
{
$cache = $this->redis->hGetAll($this->getRedisKey());
if (is_array($cache)) {
foreach ($cache as $item) {
$this->$item = $item;
}
}
}
/**
* @return \Src\order\Order
*/
public function getOrder(): Order
{
return $this->order;
}
/**
* @param \Src\order\Order $order
*/
public function setOrder(Order $order)
{
$this->order = $order;
}
}
\ No newline at end of file
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/10/11
* Time: 下午11:05
*/
namespace Src\pay_rule;
class CashPay extends Rule
{
private $startTime = '';
public function getPayAmount(): int
{
$this->startTime = $this->orderModel->out_date;
$today = time();
$startTime = strtotime($this->startTime);
$rentTime = $today - $startTime;
$rentDay = ceil($rentTime / (60 * 60));
$system = DB::table('system')->where('id', 1)->first(['free_day', 'day_money']);
if (blank($system) === true) {
abort(500, 'system config wrong');
Log::error('system config wrong');
}
$system = $system->toArray();
$this->freeDay = $system['free_day'];
$price = $system['day_money'];
if (($rentDay % 24) > 5) {
$rentDay = (int)($rentDay / 24) + 1;
} else {
$rentDay = (int)($rentDay / 24);
}
$payDay = ($rentDay - $this->freeDay) > 0 ? ($rentDay - $this->freeDay) : 0;
//todo 检查优惠券
return $price * $payDay;
}
}
\ No newline at end of file
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/10/11
* Time: 下午11:03
*/
namespace Src\pay_rule;
use App\model\OrderModel;
abstract class Rule
{
/**
* @var OrderModel
*/
protected $orderModel;
protected $freeDay = 0;
/**
* @return int
*/
public function getFreeDay(): int
{
return $this->freeDay;
}
/**
* @param int $freeDay
*/
public function setFreeDay(int $freeDay): void
{
$this->freeDay = $freeDay;
}
/**
* @param OrderModel $orderModel
*/
public function setOrderModel(OrderModel $orderModel): void
{
$this->orderModel = $orderModel;
}
public function getOrderModel(): OrderModel
{
$this->orderModel;
}
abstract public function getPayAmount(): int;
}
\ No newline at end of file
<?php
/**
* Created by PhpStorm.
* User: banice
* Date: 2018/10/12
* Time: 上午12:35
*/
namespace src\update;
use App\model\OrderModel;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;
use Lib\aliyun\api\PublishToTopic;
use Lib\aliyun\Client;
use Lib\command\OpenDoor;
use Src\battery\Battery;
use src\battery\SearchMaxQuantity;
use Src\box\Box;
use Src\box\door\Door;
use Src\order\Order;
class ReturnDoor
{
use SearchMaxQuantity;
private $outTradeNo = '';
private $batteryCode = '';
private $doorNum = 0;
private $boxNum = 0;
private $boxCode = '';
private $boxId = 0;
private $doorId = 0;
//fixme 解耦职责链
public function handle()
{
//todo 检查订单的准确性,下发打开取电舱门,web socket 推送用户还电成功
$orderModel = OrderModel::where('code', $this->outTradeNo)->first(['in_box_id', 'in_room_id', 'guest_id']);
$userId = $orderModel->guest_id;
// 更新归还电池的box_id 和 door_id door_name box_type;
$returnBox = new Box();
$returnBox->setCode($this->boxCode);
$returnDoor = new Door();
DB::table('batteries')->where('code', $this->batteryCode)->update([
'box_id' => $this->boxId,
'door_id' => $this->doorId,
'door_name' => $this->doorNum,
'box_type' => $this->boxNum,
]);
$batteryQuantity = $this->getBatteryQuantity($this->batteryCode);
//fixme 归还站点更新
DB::table('orders')->where('code', $this->outTradeNo)->update([
'in_room_id' => $this->doorId,
'in_room_name' => $this->doorNum,
'in_battery_code' => $this->batteryCode,
'in_date' => date('Y-m-d H:i:s'),
'in_power' => $batteryQuantity,
]);
$battery = new Battery();
$maxQuantityBatteryId = $battery->searchMaxQuantity($this->boxCode);
//电量最大的电池舱门
if ($maxQuantityBatteryId !== 0) {
$battery = DB::table('batteries')->where('id', $maxQuantityBatteryId)->first(['box_type', 'door_name', 'box_id', 'room_id']);
//todo 检查电池是否在当前电柜下
//todo 打开满电舱门。生成新的订单
$order = new Order();
$order->setBoxCode($this->boxCode);
$order->setUserId($userId);
$outTradeNo = time() . $userId . rand(100000, 999999);
$order->setOutTradeNo($outTradeNo);
$takeBox = new Box();
$takeDoor = new Door();
$takeBox->setId($battery->box_id);
$takeDoor->setId($battery->room_id);
$takeDoor->setName($battery->door_name);
$order->setTakeDoor($takeDoor);
$order->setTakeBox($takeBox);
$order->setBoxCode($this->boxCode);
$newOrderModel = new OrderModel();
$newOrderModel->initOrder($order);
// 下发打开舱门命令
$openTakeDoor = new OpenDoor();
$openTakeDoor->setBox($takeBox);
$openTakeDoor->setDoor($takeDoor);
$openTakeDoor->setOperationType(0);
$openTakeDoor->setOrderId($outTradeNo);
$publishToTopic = new PublishToTopic();
$publishToTopic->setMessageContent($openTakeDoor->getMessage());
$publishToTopic->setProductKey(env('MQTT_PRODUCT_KEY'));
$publishToTopic->setTopicFullName(Box::getOpenDoorTopic($this->boxCode));
$client = new Client();
$client->setApi($publishToTopic);
if ($client->request() === true) {
//todo websocket 推送消息至客户端,内容包含订单号,打开舱门和电池编号
Log::debug('send open take door success');
} else {
Log::error('send open take door fail');
//todo websocket 推送打开舱门失败,请重试
}
}
}
/**
* @return string
*/
public function getOutTradeNo(): string
{
return $this->outTradeNo;
}
/**
* @param string $outTradeNo
*/
public function setOutTradeNo(string $outTradeNo): void
{
$this->outTradeNo = $outTradeNo;
}
/**
* @return string
*/
public function getBatteryCode(): string
{
return $this->batteryCode;
}
/**
* @param string $batteryCode
*/
public function setBatteryCode(string $batteryCode): void
{
$this->batteryCode = $batteryCode;
}
/**
* @return int
*/
public function getDoorNum(): int
{
return $this->doorNum;
}
/**
* @param int $doorNum
*/
public function setDoorNum(int $doorNum): void
{
$this->doorNum = $doorNum;
}
/**
* @return int
*/
public function getBoxNum(): int
{
return $this->boxNum;
}
/**
* @param int $boxNum
*/
public function setBoxNum(int $boxNum): void
{
$this->boxNum = $boxNum;
}
/**
* @return string
*/
public function getBoxCode(): string
{
return $this->boxCode;
}
/**
* @param string $boxCode
*/
public function setBoxCode(string $boxCode): void
{
$this->boxCode = $boxCode;
}
/**
* @return int
*/
public function getBoxId(): int
{
return $this->boxId;
}
/**
* @param int $boxId
*/
public function setBoxId(int $boxId): void
{
$this->boxId = $boxId;
}
/**
* @return int
*/
public function getDoorId(): int
{
return $this->doorId;
}
/**
* @param int $doorId
*/
public function setDoorId(int $doorId): void
{
$this->doorId = $doorId;
}
}
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment