基于PHP swoole扩展短信通道cmpp接入

<?php /** * cmpp服务 - 联通 - 长连接 */ error_reporting(1); ini_set('display_errors','on'); define('CMPP_CONNECT', 0x00000001); // 请求连接 define('CMPP_CONNECT_RESP', 0x80000001); // 请求连接应答 define('CMPP_TERMINATE', 0x00000002); // 终止连接 define('CMPP_TERMINATE_RESP', 0x80000002); // 终止连接应答 define('CMPP_SUBMIT', 0x00000004); // 提交短信 define('CMPP_SUBMIT_RESP', 0x80000004); // 提交短信应答 define('CMPP_DELIVER', 0x00000005); // 短信下发 define('CMPP_DELIVER_RESP', 0x80000005); // 下发短信应答 define('CMPP_ACTIVE_TEST', 0x00000008); // 链路检测|激活测试 define('CMPP_ACTIVE_TEST_RESP', 0x80000008); // 链路检测|激活测试应答 class Apilib_Server_Cucc { protected $clients; protected $backends; protected $server; protected $serv; protected $fd; protected $sequence_id = 1; protected $redis_client; protected $sp_code; // 企业代码 protected $password; // 密码 protected $ip; // IP protected $port; // 端口 protected $src_id; // 接入号 function __construct($Config = NULL) { $channel = Channel_DsModel::get4cmpp($stype = Channel_InfoModel::TYPE_NORMAL, $ctype = Channel_InfoModel::CTYPE_CUCC); $this->ip = $channel['ip']; $this->port = $channel['port']; $this->sp_code = $channel['sp_code']; $this->password = $channel['password']; $this->src_id = $channel['src_id']; $this->serv = new swoole_server("0.0.0.0", 9784); $this->serv->set(array( 'timeout' => 1, //select and epoll_wait timeout. 'daemonize' => 1, //以守护进程执行 'reactor_num' => 1, //线程数 'worker_num' => 1, //进程数 'backlog' => 128, //Listen队列长度 'max_conn' => 10000, 'max_request' => 3000, 'task_max_request'=>3000, 'dispatch_mode' => 2, 'debug_mode'=> 0, 'log_file' => '/tmp/server_cucc.log', )); $this->serv->on('workerstart', array($this, 'onStart')); $this->serv->on('Receive', array($this, 'onReceive')); $this->serv->on('close', array($this, 'onClose')); $this->serv->start(); } /** * 获得redis连接 */ private function redis_conn() { if (!$this->redis_client || !$this->redis_client->pingCheck()) { $peristent = true; $this->redis_client = Comm_Redis::r(Comm_Redis::RDQ, $peristent); } return $this->redis_client; } /* * 登录成功后每隔两分钟发一次心跳 */ private function CMPP_CONNECT_RESP($pdu){ $format = "CStatus/a16Auth/CVersion"; $data = unpack($format, $pdu); $status = intval($data['Status']); echo "第一步-请求连接应答："; if($status !== 0){ return false; } $client = $this->clients['sys']['socket']; $this->sequence_id = $this->sequence_id + 1; $test = $this->sequence_id; $this->serv->tick(125000, function()use ($client, $test){ $header = pack('NNN',12,CMPP_ACTIVE_TEST,$test); $client->send($header); }); } /* * 登录成功后每隔两分钟发一次心跳 */ private function CMPP_CONNECT_SRV_RESP($data, $serv, $fd){ $format = "a6Source_Addr/a16AuthenticatorSource/CVersion/NTimestamp"; $pdu = substr($data, 12); $pdu_data = unpack($format, $pdu); //$this->_log("发送连接响应pdu:".print_r($pdu_data, true)); $pdu_data['AuthenticatorSource'] = self::pduord($pdu_data['AuthenticatorSource']); //$this->_log("CMPP_CONNECT PDU: " . self::pduord($pdu_data), ADC_LOG_DEBUG); //$this->_log("CMPP_CONNECT SPEC: " . $format, ADC_LOG_DEBUG); //$this->_log("CMPP_CONNECT DATA: " . print_r($pdu_data, true), ADC_LOG_DEBUG); //unset($debug_data); $market = Merchant_CmppModel::gets($pdu_data['Source_Addr']); //服务端AuthenticatorSource $sdata= []; $sdata['Shared_Secret'] = $market[$pdu_data['Source_Addr']]['password']; $sdata['Timestamp'] = sprintf("%010d", $pdu_data['Timestamp']); $AuthenticatorSource_ori = $pdu_data['Source_Addr'] . pack('a9', '') . $sdata['Shared_Secret'] . $sdata['Timestamp']; $AuthenticatorSource = md5($AuthenticatorSource_ori, true); $sdata['AuthenticatorSource'] = self::pduord($AuthenticatorSource); //$this->_log("CMPP_CONNECT SDATA: " . print_r($sdata, true), ADC_LOG_DEBUG); $conn_list = implode(",",$serv->getClientList(0,10)); $stats_info = $serv->stats(); extract(unpack("Nsequence_id", substr($data, 8, 4))); $conn_num = $stats_info['connection_num']; $accept_count = $stats_info['accept_count']; $close_count = $stats_info['close_count']; $tasking_num = $stats_info['tasking_num']; Helper_Log::writeApplog("connection_cucc", "$conn_num\t$accept_count\t$close_count\t$tasking_num\t$sequence_id\t$conn_list\r\n"); if($stats_info['connection_num'] > Comm_Config::get("sms.basic.limit_conn")) { $Status = 110; $Auth = "1234567890123456"; $Version = 0; $format = "Ca16C"; $data = pack($format, $Status, $Auth, $Version); $command_id = CMPP_CONNECT_RESP; $length = strlen($data) + 12; $header = pack("NNN", $length, $command_id, $sequence_id); $serv->send($fd, $header.$data); $serv->close($fd); } $client_info = $serv->getClientInfo($fd); if (($pdu_data['AuthenticatorSource'] == $sdata['AuthenticatorSource'] && $market[$pdu_data['Source_Addr']]['status'] == 1) || $client_info['remote_ip'] == "127.0.0.1") { $this->_log("auth succ"); //连接应答 $Status = 0; $Auth = "1234567890123456"; $Version = 0x20; $conn_res = []; $conn_res['Status'] = $Status; $conn_res['Auth'] = $Auth; $conn_res['Version'] = $Version; //$this->_log("发送应答报文内容:".print_r($conn_res, true)); $format = "Ca16C"; $data = pack($format, $Status, $Auth, $Version); $command_id = CMPP_CONNECT_RESP; $length = strlen($data) + 12; //$this->_log("发送应答报文header:".print_r([$length, $command_id, 1], true)); $header = pack("NNN", $length, $command_id, $sequence_id); $remote_ip = $client_info['remote_ip']; Helper_Log::writeApplog("connection_cucc", "$remote_ip\t$sequence_id\r\n"); //$this->_log("应答对象:".print_r($fd, true)); $serv->send($fd, $header.$data); } else { $this->_log("auth fail"); $serv->close($fd); } } /* * 客户端收到心跳包后回的响应包 */ private function CMPP_ACTIVE_SRV_RESP($data, $fd, $serv){ extract(unpack("Nsequence_id", substr($data, 8, 4))); $format = "C"; $body = pack($format, ''); echo "第四步-SRV-激活测试应答:".$sequence_id; $client_info = $serv->getClientInfo($fd); $stats = $serv->stats(); $ip = $client_info['remote_ip']; $port = $client_info['remote_port']; $connect_time = $client_info['connect_time']; $last_time = $client_info['last_time']; $start_time = $stats['start_time']; $connection_num = $stats['connection_num']; $accept_count = $stats['accept_count']; $close_count = $stats['close_count'];