czapp-api/Home/Lib/Action/NoticeAction.class.php

<?php



class NoticeAction extends Action {
			


 
	public  function jgpush(  ){
		$app_key = '58e15e345e0c21b7e638e186';
		$master_secret = 'a91d0b1c1fa1b327db832506';
		$client = new \JPush\Client($app_key, $master_secret);
		$push_payload = $client->push()
		    ->setPlatform('all')
		    ->addAllAudience()
		    ->setNotificationAlert('车辆异动告警');
		try {
		    $response = $push_payload->send();
		    print_r($response);
		} catch (\JPush\Exceptions\APIConnectionException $e) {
		    // try something here
		    print $e;
		} catch (\JPush\Exceptions\APIRequestException $e) {
		    // try something here
		    print $e;
		}
	}
	
 
	public  function pushFromKafka(  ){
		$conf = new RdKafka\Conf();
		// Set a rebalance callback to log partition assignments (optional)（当有新的消费进程加入或者退出消费组时，kafka 会自动重新分配分区给消费者进程，这里注册了一个回调函数，当分区被重新分配时触发）
		$conf->setRebalanceCb(function (RdKafka\KafkaConsumer $kafka, $err, array $partitions = null) {
		    switch ($err) {
		        case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
		            echo "Assign: ";
		            var_dump($partitions);
		            $kafka->assign($partitions);
		            break;
		
		         case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS:
		             echo "Revoke: ";
		             var_dump($partitions);
		             $kafka->assign(NULL);
		             break;
		
		         default:
		            throw new \Exception($err);
		    }
		});
		
		// Configure the group.id. All consumer with the same group.id will consume（ 配置groud.id 具有相同 group.id 的consumer 将会处理不同分区的消息，所以同一个组内的消费者数量如果订阅了一个topic， 那么消费者进程的数量多于这个topic 分区的数量是没有意义的。）
		// different partitions.
		$conf->set('group.id', 'myConsumerGroup');
		
		// Initial list of Kafka brokers（添加 kafka集群服务器地址）
		$conf->set('metadata.broker.list', '127.0.0.1');
		
		$topicConf = new RdKafka\TopicConf();
		
		// Set where to start consuming messages when there is no initial offset in
		// offset store or the desired offset is out of range.
		// 'smallest': start from the beginning
		$topicConf->set('auto.offset.reset', 'smallest');
		
		// Set the configuration to use for subscribed/assigned topics
		$conf->setDefaultTopicConf($topicConf);
		
		$consumer = new RdKafka\KafkaConsumer($conf);
		
		// 消费者订阅
		$consumer->subscribe(['gps_alarm_msg_queue']);
		
		while (true) {
		    $message = $consumer->consume(120*1000);
		    switch ($message->err) {
		        case RD_KAFKA_RESP_ERR_NO_ERROR:
		           $msg_data = json_decode($message->payload,true);
						if($msg_data){
						   // 使用极光推送消息
							$jpush_client = new \JPush\Client( C('JPUSH_APP_KEY'), C('JPUSH_MASTER_SECRET') );
							// 电子围栏告警，上锁车辆异动告警，低电量告警，被盗告警，广播消息推送
							$this->jpushMsg( $jpush_client, $message->payload );
						}
		            break;
		        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
		            echo "No more messages; will wait for more\n";
		            break;
		        case RD_KAFKA_RESP_ERR__TIMED_OUT:
		            echo "Timed out\n";
		            break;
		        default:
		            throw new \Exception($message->errstr(), $message->err);
		            break;
		    }
		}
	}
	
 
	public  function jpushMsg( $client, $msg_data ){
		/*
		$msg_data = '{
			  "type":2,
			  "title":"被盗告警",
			  "content":"车被偷了，赶紧去找",
			  "device_number":"FFFFFF123122"
			}';
		$msg_data = '{
		  "type":9,
		  "title":"群推消息",
		  "content":"这是一个广播"
		}'; 
		*/ 
		
		$single_push_type = array(
			C('LOCK_VEHICLE_ALARM'),   // 1-锁车告警
			C('STOLEN_ALARM'),       // 2-被盗告警
			C('FENCE_ALARM'),        // 3-电子围栏告警
			C('LOWWER_BATTERY_ALARM'),  // 4-低电量告警
		);
		$msg_data = json_decode($msg_data,true);
		if(!$msg_data){
			echo 'invalid message data format!'.$msg_data . PHP_EOL;
			return;
		}
		// 通过传过来的车牌查出JgClientRegistrationId
		if( empty($client) ){
			$client = new \JPush\Client( C('JPUSH_APP_KEY'), C('JPUSH_MASTER_SECRET') );
		}
		if( in_array($msg_data['type'],$single_push_type) && $msg_data['device_number'] ){
			$where = array('DeviceNumber|GpsDeviceNumber' => $msg_data['device_number']);
			$fields= 'CityId,LicensePlate,FullName,JgClientRegistrationId';
			
			if( !S( 'jpush_vinfo_'.$msg_data['device_number'] ) ){ //如果不存在，则缓存
				$vehicle_info = M('jms_vehicle')->field($fields)->where($where)->find();
				S( 'jpush_vinfo_'.$msg_data['device_number'], $vehicle_info, 60 ); //缓存1分钟
			}
			$vehicle_info = S( 'jpush_vinfo_'.$msg_data['device_number'] ); //从缓存取
			
		}
		// 推送平台
		$platform = array('ios', 'android');
		// 通知栏显示内容
		$alert = $msg_data['content'];
		// 推送android
		$android_notification = array(
		    'title' => $msg_data['title'],
		    'builder_id' => 2, //通知栏样式
		);
		// 推送ios
		$ios_notification = array(
		    'sound' => 'default',
		    'badge' => '+1',
		    'content-available' => true
		);
		// 可选项
		$options = array(
		    'sendno' => 100,
		    'time_to_live' => 100,
		);
		// 组装推送
		$cid = $client->push()->getCid();
		//var_dump($cid['body']['cidlist'][0]);
		$push_payload = $client->push()
			 ->setCid($cid['body']['cidlist'][0])
		    ->setPlatform( $platform )
		    ->iosNotification($alert, $ios_notification)
		    ->androidNotification($alert, $android_notification)
		    ->options($options);
		
		if( $msg_data['type'] == C('BROADCASTING') ){ // 9-广播
		   $push_payload->addAllAudience();
		}elseif( isset($vehicle_info) && $vehicle_info ){ // 单推
			$msg_data['vehicle_info'] = $vehicle_info;
			$push_payload->addRegistrationId($vehicle_info['JgClientRegistrationId']);
		}else{
			//未知类型
			var_dump('出错了：'.$msg_data);
			return;
		}
		try {
		    $response = $push_payload->send();
			 // 保存日志
			 $msg_data['response'] = array(
			 			'code' => $response['http_code'],
				 		'resp_msg' => 'ok',
			 );
			 $this->saveLog($msg_data);
		    print_r($response);
		} catch (\JPush\Exceptions\APIConnectionException $e) {
		    // try something here
			$msg_data['response'] = array(
			 			'code' => $e->getHttpCode(),
				 		'resp_msg' => substr($e->__toString(), strpos($e->__toString(),'[')),
			 );
			$this->saveLog($msg_data);
			print $e;
		} catch (\JPush\Exceptions\APIRequestException $e) {
		    // try something here
			$msg_data['response'] = array(
			 			'code' => $e->getHttpCode(),
				 		'resp_msg' => substr($e->__toString(), strpos($e->__toString(),'[')),
			 );
			$this->saveLog($msg_data);
			print $e;
		}
	}
	
 
	public  function saveLog( $msg_data ){
		$fdls_app_message = M('jms_baojing_message');
		
		$log_data = array(
			'ID' => create_guid(),
			'Type' => $msg_data['type'],
			'Title' => $msg_data['title'],
			'Comment' => $msg_data['content'],
			'SendStatus' => $msg_data['response']['code'],
			'AddTime' => date('Y-m-d H:i:s'),
			'RespMsg' => $msg_data['response']['resp_msg'],
		);
		if( $msg_data['type'] != C('BROADCASTING') ){ // 非广播
			$log_data['CityId'] 	  = $msg_data['vehicle_info']['CityId'];
			$log_data['DeviceNumber'] = $msg_data['device_number'];
			$log_data['FullName']	  = $msg_data['vehicle_info']['FullName'];
			$log_data['LicensePlate'] = $msg_data['vehicle_info']['LicensePlate'];
		}
		
		$result = $fdls_app_message->createAdd($log_data);
		return $result;
		
	}
	
 
	public  function kafkaProducer( $msg_data ){
		if (!extension_loaded('rdkafka')){
			echo 'rdkafka extension is not installed!!'.PHP_EOL;
			return false;
		}
		/********************* 初始化生产者配置项 start **************************/
		//考勤记录分析结果生产者
		$rk = new RdKafka\Producer();
		$rk->setLogLevel(LOG_DEBUG);
		$rk->addBrokers("127.0.0.1");
		$start = microtime(true);
		$topic = $rk->newTopic('gps_alarm_msg_queue');
		/********************* 初始化生产者配置项 end **************************/
		if( empty($msg_data) ){
			var_dump($msg_data);
			return;
		}
		if( is_array($msg_data) || is_object($msg_data) ){ // 对象、数组转换
			$msg_data = json_encode($msg_data);
		}
		$topic->produce(RD_KAFKA_PARTITION_UA, 0, $msg_data);
		$rk->poll(0);
		
	}
	
 
	public  function mockProduce(  ){
		$msg_data = '{
		  "type":9,
		  "title":"群推消息",
		  "content":"这是一个广播"
		}';
		$this->kafkaProducer($msg_data);
	}
	

}