<commit_before><?php
/**
 * Copyright 2017 <NAME>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

/**
 * Created by Liming
 * Date: 2016/12/5
 * Time: 14:52
 */
use Constant\DELIVERY_MESSAGE;
use Database\Result;
use Workerman\Connection\AsyncTcpConnection;
use Workerman\Connection\TcpConnection;
use Workerman\Lib\Timer;
use Workerman\Worker;

require __DIR__.'/../Channel/src/Client.php';
require __DIR__.'/../Channel/src/Server.php';
require __DIR__.'/message.php';

//Workerman 通讯服务
$channel_server = new Channel\Server(CONFIG['websocket']['channel']['listen'], CONFIG['websocket']['channel']['port']);

//心跳包间隔（秒）
define('HEARTBEAT_TIME', 300);
$worker = new Worker('websocket://'.CONFIG['websocket']['listen']);
//同时服务进程数
$worker->count = CONFIG['websocket']['process'];

/**
 * 启动服务
 *
 * @param Worker $worker
 */
$worker->onWorkerStart = function(Worker $worker) use ($MESSAGE_TYPE, $MESSAGE_CODE) {
    //任务队列
    $worker->process_pool = [];
    //心跳检测
    Timer::add(60, function() use ($worker) {
        $time_now = time();
        foreach($worker->connections as $connection) {
            if(empty($connection->lastMessageTime)) {
                $connection->lastMessageTime = $time_now;
                continue;
            }
            if($time_now - $connection->lastMessageTime > HEARTBEAT_TIME) {
                $connection->close();
            }
        }
    });
    //任务处理
    Timer::add(5, function() use ($worker, $MESSAGE_TYPE, $MESSAGE_CODE) {
        $count = count($worker->process_pool);
        if($count > 0) {
            //查询可用服务数
            $task = new AsyncTcpConnection('text://'.CONFIG['websocket']['delivery']);
            $task->onMessage = function(AsyncTcpConnection $task, string $message) use ($worker, $count, $MESSAGE_TYPE, $MESSAGE_CODE) {
                $task->close();
                //创建服务请求
                $count = min($count, intval($message));
                for($i = 0; $i < $count; ++$i) {
                    $task = new AsyncTcpConnection('text://'.CONFIG['websocket']['delivery']);
                    $task->task = array_shift($worker->process_pool);
                    $task->onMessage = function(AsyncTcpConnection $task, string $message) use ($worker, $MESSAGE_TYPE, $MESSAGE_CODE) {
                        $message = json_decode($message);
                        if($message->code == DELIVERY_MESSAGE::REQUEST_SUCCEED) {
                            //请求服务成功，发送任务
                            $task->send(json_encode([
                                'type' => DELIVERY_MESSAGE::JUDGE,
                                'data' => $task->task->judge_info
                            ]));
                            //向用户发送开始消息
                            Channel\Client::publish('message', [
                                'user'    => (string)$task->task->uid,
                                'message' => [
                                    'code'    => $MESSAGE_CODE->StartJudging,
                                    'type'    => $MESSAGE_TYPE->Judge,
                                    'message' => 'Start Judging',
                                    'id'      => (string)$task->task->rid,
                                    '_t'      => timestamp()
                                ]
                            ]);
                        } elseif($message->code == DELIVERY_MESSAGE::JUDGE_SUCCEED) {
                            $task->close();
                            //任务完成
                            Result::getInstance()->updateResult($task->task->rid, $message->result);
                            //向用户发送结果
                            $ret = [
                                'code'    => $message->result,
                                'type'    => $MESSAGE_TYPE->JudgeResult,
                                'message' => 'Judge Result',
                                'id'      => (string)$task->task->rid,
                                '_t'      => timestamp()
                            ];
                            if(isset($message->info)) {
                                $ret['info'] = $message->info;
                            }
                            Channel\Client::publish('message', [
                                'user'    => (string)$task->task->uid,
                                'message' => $ret
                            ]);
                        } elseif($message->code == DELIVERY_MESSAGE::REQUEST_FAILED) {
                            $task->close();
                            //请求任务失败，回到队伍前排继续
                            array_unshift($worker->process_pool, $task->task);
                        }
                    };
                    $task->send(json_encode(['type' => DELIVERY_MESSAGE::REQUEST]));
                    $task->connect();
                }
            };
            $task->send(json_encode(['type' => DELIVERY_MESSAGE::AVAILABLE]));
            $task->connect();<commit_msg>Bug fix: Should be an array rather than a stdClass object.🏈<commit_after><?php
/**
 * Copyright 2017 <NAME>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

/**
 * Created by Liming
 * Date: 2016/12/5
 * Time: 14:52
 */
use Constant\DELIVERY_MESSAGE;
use Database\Result;
use Workerman\Connection\AsyncTcpConnection;
use Workerman\Connection\TcpConnection;
use Workerman\Lib\Timer;
use Workerman\Worker;

require __DIR__.'/../Channel/src/Client.php';
require __DIR__.'/../Channel/src/Server.php';
require __DIR__.'/message.php';

//Workerman 通讯服务
$channel_server = new Channel\Server(CONFIG['websocket']['channel']['listen'], CONFIG['websocket']['channel']['port']);

//心跳包间隔（秒）
define('HEARTBEAT_TIME', 300);
$worker = new Worker('websocket://'.CONFIG['websocket']['listen']);
//同时服务进程数
$worker->count = CONFIG['websocket']['process'];

/**
 * 启动服务
 *
 * @param Worker $worker
 */
$worker->onWorkerStart = function(Worker $worker) use ($MESSAGE_TYPE, $MESSAGE_CODE) {
    //任务队列
    $worker->process_pool = [];
    //心跳检测
    Timer::add(60, function() use ($worker) {
        $time_now = time();
        foreach($worker->connections as $connection) {
            if(empty($connection->lastMessageTime)) {
                $connection->lastMessageTime = $time_now;
                continue;
            }
            if($time_now - $connection->lastMessageTime > HEARTBEAT_TIME) {
                $connection->close();
            }
        }
    });
    //任务处理
    Timer::add(5, function() use ($worker, $MESSAGE_TYPE, $MESSAGE_CODE) {
        $count = count($worker->process_pool);
        if($count > 0) {
            //查询可用服务数
            $task = new AsyncTcpConnection('text://'.CONFIG['websocket']['delivery']);
            $task->onMessage = function(AsyncTcpConnection $task, string $message) use ($worker, $count, $MESSAGE_TYPE, $MESSAGE_CODE) {
                $task->close();
                //创建服务请求
                $count = min($count, intval($message));
                for($i = 0; $i < $count; ++$i) {
                    $task = new AsyncTcpConnection('text://'.CONFIG['websocket']['delivery']);
                    $task->task = array_shift($worker->process_pool);
                    $task->onMessage = function(AsyncTcpConnection $task, string $message) use ($worker, $MESSAGE_TYPE, $MESSAGE_CODE) {
                        $message = json_decode($message);
                        if($message->code == DELIVERY_MESSAGE::REQUEST_SUCCEED) {
                            //请求服务成功，发送任务
                            $task->send(json_encode([
                                'type' => DELIVERY_MESSAGE::JUDGE,
                                'data' => $task->task['judge_info']
                            ]));
                            //向用户发送开始消息
                            Channel\Client::publish('message', [
                                'user'    => (string)$task->task['uid'],
                                'message' => [
                                    'code'    => $MESSAGE_CODE->StartJudging,
                                    'type'    => $MESSAGE_TYPE->Judge,
                                    'message' => 'Start Judging',
                                    'id'      => (string)$task->task['rid'],
                                    '_t'      => timestamp()
                                ]
                            ]);
                        } elseif($message->code == DELIVERY_MESSAGE::JUDGE_SUCCEED) {
                            $task->close();
                            //任务完成
                            Result::getInstance()->updateResult($task->task['rid'], $message->result);
                            //向用户发送结果
                            $ret = [
                                'code'    => $message->result,
                                'type'    => $MESSAGE_TYPE->JudgeResult,
                                'message' => 'Judge Result',
                                'id'      => (string)$task->task['rid'],
                                '_t'      => timestamp()
                            ];
                            if(isset($message->info)) {
                                $ret['info'] = $message->info;
                            }
                            Channel\Client::publish('message', [
                                'user'    => (string)$task->task['uid'],
                                'message' => $ret
                            ]);
                        } elseif($message->code == DELIVERY_MESSAGE::REQUEST_FAILED) {
                            $task->close();
                            //请求任务失败，回到队伍前排继续
                            array_unshift($worker->process_pool, $task->task);
                        }
                    };
                    $task->send(json_encode(['type' => DELIVERY_MESSAGE::REQUEST]));
                    $task->connect();
                }
            };
            $task->send(json_encode(['type' => DELIVERY_MESSAGE::AVAILABLE]));
            $task->connect();
{'fraction_non_alphanumeric': 0.1377959927140255, 'fraction_numerical': 0.004918032786885246, 'mean_word_length': 1.4326539654408508, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 131, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3496834', 'n_tokens_mistral': 2972, 'n_tokens_neox': 2867, 'n_words': 541}