diff --git a/README.md b/README.md index 9d31349..cd5e60a 100644 --- a/README.md +++ b/README.md @@ -33,12 +33,14 @@ Car는 차량 모뎀/게이트웨이에서 받은 상태를 수집해 DB에 저 - `monitor.php`: 모니터링 화면과 AJAX 응답 - `common.php`: 외부 secret 로드와 공통 DB/API 함수 - `collector_se.php`: 상태 수집 CLI/cron +- `tcp_worker.php`: TCP 큐 처리와 늦은 사용량 정산 worker +- `car-tcp-worker.service`: systemd service 예시 - `sw.js`: service worker - `assets/`: 아이콘과 정적 자산 ## 데이터/저장소 -- Car DB tables: 차량 상태, 로그, 차트용 기록 +- Car DB tables: 차량 상태, 로그, 차트용 기록, TCP 요청 큐 - 사용량 수집값과 월별 보정 기준 - `/home/seo/secret/car.php`: TCP, DB, token, 허용 IP 설정 @@ -48,13 +50,15 @@ Car는 차량 모뎀/게이트웨이에서 받은 상태를 수집해 DB에 저 2. `monitor.php`는 상태 AJAX와 사용량 AJAX를 분리해 갱신합니다. 3. 제어 요청은 token/IP 정책을 검증합니다. 4. 명령 코드를 허용 목록과 대조합니다. -5. TCP 게이트웨이로 전송하고 결과와 실패 reason을 반환합니다. +5. 요청을 `car_tcp_request_queue`에 저장합니다. +6. `car-tcp-worker.service`가 큐를 처리하며 TCP 게이트웨이로 전송합니다. +7. API는 큐 상태를 4.95초까지만 기다린 뒤 결과 또는 timeout을 반환합니다. ## TCP timeout과 사용량 정산 차량 명령 성공/실패 판단은 UI 안정성을 위해 `TCP_TOTAL_TIMEOUT` 기본값인 4.95초를 사용합니다. 이 시간 안에 정상 응답이 없으면 API는 기존처럼 실패를 반환하므로 자동화와 UI는 즉시 재시도 여부를 판단할 수 있습니다. -단, 실제 모뎀이 4.95초 이후에 늦게 응답할 수 있으므로 사용량 기록은 별도로 정산합니다. 4.95초까지 정상 응답이 없고 송신이 완료된 요청은 `pcntl_fork()`로 자식 프로세스를 만들고, 같은 TCP 소켓을 최대 `TCP_USAGE_SETTLEMENT_TIMEOUT` 기본 60초까지 더 관찰합니다. +단, 실제 모뎀이 4.95초 이후에 늦게 응답할 수 있으므로 사용량 기록은 별도로 정산합니다. API는 요청을 DB 큐에 넣고 4.95초까지만 기다립니다. `car-tcp-worker.service`는 해당 요청의 TCP 연결을 직접 열고, UI timeout 이후에도 같은 소켓을 최대 `TCP_USAGE_SETTLEMENT_TIMEOUT` 기본 60초까지 더 관찰합니다. - 60초 안에 정상 응답이 완성되면 `late_response_after_ui_timeout`으로 송신량과 수신량을 함께 기록합니다. - 60초까지 응답이 없으면 `late_total_timeout_60s_sent_only`로 송신 패킷만 사용량에 반영합니다. @@ -63,12 +67,26 @@ Car는 차량 모뎀/게이트웨이에서 받은 상태를 수집해 DB에 저 이 구조에서 `request_id`는 UI 응답과 늦은 사용량 정산 기록을 연결하는 기준입니다. 즉 명령 재시도 로직은 4.95초 기준을 유지하고, 데이터 사용량은 최대 60초까지 늦은 수신 패킷을 반영합니다. +## systemd worker + +설치 파일은 `car-tcp-worker.service`입니다. + +```bash +cp /var/www/seoul/car/car-tcp-worker.service /etc/systemd/system/car-tcp-worker.service +systemctl daemon-reload +systemctl enable --now car-tcp-worker.service +systemctl status car-tcp-worker.service +``` + +큐 테이블은 코드에서 자동 생성합니다. Worker가 중지되면 API 요청은 큐에 쌓이지만 4.95초 안에 처리되지 않아 실패로 반환될 수 있습니다. 안전을 위해 worker가 뒤늦게 오래된 queued 요청을 발견하면 실제 TCP 송신 없이 `queue_expired_before_send`로 만료 처리합니다. + ## 주요 함수/모듈 - `common.php`: secret 로드와 공통 DB/API 함수 - `api.php`: 상태 조회와 명령 전송 - `monitor.php`: 화면 렌더링과 AJAX 응답 - `collector_se.php`: 상태 수집과 중복 실행 방지 +- `tcp_worker.php`: TCP 요청 큐 처리와 60초 사용량 정산 ## 보안 @@ -83,3 +101,4 @@ Car는 차량 모뎀/게이트웨이에서 받은 상태를 수집해 DB에 저 - 명령별 rate limit과 감사 로그를 유지합니다. - 통신사 기준 데이터 사용량 보정값을 주기적으로 확인합니다. - 4.95초 timeout 직후에는 사용량 기록이 즉시 생기지 않을 수 있으며, 최대 60초 뒤 늦은 정산 행이 추가될 수 있습니다. +- `systemctl status car-tcp-worker.service`와 `journalctl -u car-tcp-worker.service`로 worker 상태를 확인합니다. diff --git a/car-tcp-worker.service b/car-tcp-worker.service new file mode 100644 index 0000000..fe07dd9 --- /dev/null +++ b/car-tcp-worker.service @@ -0,0 +1,16 @@ +[Unit] +Description=Car TCP queue worker +After=network-online.target mariadb.service mysql.service +Wants=network-online.target + +[Service] +Type=simple +WorkingDirectory=/var/www/seoul/car +ExecStart=/usr/bin/php /var/www/seoul/car/tcp_worker.php +Restart=always +RestartSec=2 +User=root +Group=root + +[Install] +WantedBy=multi-user.target diff --git a/common.php b/common.php index d20e0a2..9744535 100644 --- a/common.php +++ b/common.php @@ -158,50 +158,300 @@ function record_tcp_usage( ); } -function finalize_late_tcp_usage( - $fp, - string $cmd, - string $usageSource, - ?string $requestId, - int $sentBytes, - string $initialResponse, - float $deadline -): void { - ignore_user_abort(true); - @set_time_limit((int)ceil(max(5.0, TCP_USAGE_SETTLEMENT_TIMEOUT + 5.0))); +function tcp_request_payload(string $cmd): string +{ + return json_encode([ + 'type' => 'R', + 'type_sub' => 'car_controll', + 'data' => [ + 'command' => '+SCMD=' . MODEM . '/C:' . $cmd, + 'modem' => MODEM, + 'user' => USER, + 'uid' => UID, + 'type' => TYPE, + ], + ], JSON_UNESCAPED_SLASHES); +} - $response = $initialResponse; - $tcpError = ''; +function db_ensure_tcp_queue_schema(PDO $pdo): void +{ + static $done = false; + if ($done) { + return; + } - while (microtime(true) < $deadline) { - $read = [$fp]; - $w = null; - $e = null; + $pdo->exec("CREATE TABLE IF NOT EXISTS car_tcp_request_queue ( + id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, + request_id VARCHAR(64) NOT NULL UNIQUE, + car_id VARCHAR(20) NOT NULL DEFAULT 'car', + created_at DATETIME NOT NULL, + updated_at DATETIME NOT NULL, + source VARCHAR(20) NOT NULL, + cmd VARCHAR(10) NOT NULL, + status VARCHAR(32) NOT NULL, + request_payload MEDIUMTEXT NOT NULL, + sent_bytes INT NOT NULL DEFAULT 0, + received_bytes INT NOT NULL DEFAULT 0, + total_bytes INT NOT NULL DEFAULT 0, + tcp_ok TINYINT NULL, + tcp_error VARCHAR(100) NULL, + raw_full MEDIUMTEXT NULL, + connect_ms INT NOT NULL DEFAULT 0, + read_ms INT NOT NULL DEFAULT 0, + tcp_ms INT NOT NULL DEFAULT 0, + ui_deadline_ts DOUBLE NOT NULL DEFAULT 0, + usage_deadline_ts DOUBLE NOT NULL DEFAULT 0, + worker_pid INT NULL, + KEY idx_status_id (status, id), + KEY idx_request_id (request_id) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4"); - $remain = $deadline - microtime(true); - if ($remain <= 0) { - break; + $done = true; +} + +function tcp_queue_now(): string +{ + return date('Y-m-d H:i:s'); +} + +function tcp_queue_update(PDO $pdo, string $requestId, array $values): void +{ + if (!$values) { + return; + } + + $values['updated_at'] = tcp_queue_now(); + $sets = []; + $params = [':request_id' => $requestId]; + + foreach ($values as $key => $value) { + $param = ':' . $key; + $sets[] = "$key = $param"; + $params[$param] = $value; + } + + $stmt = $pdo->prepare("UPDATE car_tcp_request_queue SET " . implode(', ', $sets) . " WHERE request_id = :request_id"); + $stmt->execute($params); +} + +function tcp_queue_insert(string $cmd, string $usageSource, string $requestId): array +{ + $pdo = db(); + db_ensure_tcp_queue_schema($pdo); + + $now = microtime(true); + $payload = tcp_request_payload($cmd); + + $stmt = $pdo->prepare("INSERT INTO car_tcp_request_queue ( + request_id, car_id, created_at, updated_at, source, cmd, status, + request_payload, ui_deadline_ts, usage_deadline_ts + ) VALUES ( + :request_id, 'car', :created_at, :updated_at, :source, :cmd, 'queued', + :request_payload, :ui_deadline_ts, :usage_deadline_ts + )"); + $stmt->execute([ + ':request_id' => $requestId, + ':created_at' => tcp_queue_now(), + ':updated_at' => tcp_queue_now(), + ':source' => substr($usageSource, 0, 20), + ':cmd' => substr($cmd, 0, 10), + ':request_payload' => $payload, + ':ui_deadline_ts' => $now + TCP_TOTAL_TIMEOUT, + ':usage_deadline_ts' => $now + TCP_USAGE_SETTLEMENT_TIMEOUT, + ]); + + return [ + 'request_id' => $requestId, + 'cmd' => $cmd, + 'source' => $usageSource, + 'request_payload' => $payload, + 'ui_deadline_ts' => $now + TCP_TOTAL_TIMEOUT, + 'usage_deadline_ts' => $now + TCP_USAGE_SETTLEMENT_TIMEOUT, + ]; +} + +function tcp_queue_get(string $requestId): ?array +{ + $pdo = db(); + db_ensure_tcp_queue_schema($pdo); + + $stmt = $pdo->prepare("SELECT * FROM car_tcp_request_queue WHERE request_id = :request_id LIMIT 1"); + $stmt->execute([':request_id' => $requestId]); + $row = $stmt->fetch(); + return $row ?: null; +} + +function tcp_queue_claim_next(PDO $pdo): ?array +{ + db_ensure_tcp_queue_schema($pdo); + + $pdo->beginTransaction(); + try { + $stmt = $pdo->query("SELECT * FROM car_tcp_request_queue WHERE status = 'queued' ORDER BY id ASC LIMIT 1 FOR UPDATE"); + $row = $stmt->fetch(); + if (!$row) { + $pdo->commit(); + return null; } - $wait = min($remain, 1.0); + $update = $pdo->prepare("UPDATE car_tcp_request_queue SET status = 'processing', updated_at = :updated_at, worker_pid = :worker_pid WHERE request_id = :request_id"); + $update->execute([ + ':updated_at' => tcp_queue_now(), + ':worker_pid' => getmypid() ?: null, + ':request_id' => $row['request_id'], + ]); + $pdo->commit(); + + $row['status'] = 'processing'; + $row['worker_pid'] = getmypid() ?: null; + return $row; + } catch (Throwable $e) { + if ($pdo->inTransaction()) { + $pdo->rollBack(); + } + throw $e; + } +} + +function tcp_queue_record_usage_and_status( + PDO $pdo, + array $job, + string $status, + int $sentBytes, + int $receivedBytes, + bool $tcpOk, + string $tcpError, + string $rawFull = '', + int $connectMs = 0, + int $readMs = 0 +): void { + $requestId = (string)$job['request_id']; + $source = (string)$job['source']; + $cmd = (string)$job['cmd']; + + db_insert_data_usage($pdo, $source, $cmd, $sentBytes, $receivedBytes, $tcpOk, $tcpError, $requestId); + + tcp_queue_update($pdo, $requestId, [ + 'status' => $status, + 'sent_bytes' => max(0, $sentBytes), + 'received_bytes' => max(0, $receivedBytes), + 'total_bytes' => max(0, $sentBytes + $receivedBytes), + 'tcp_ok' => $tcpOk ? 1 : 0, + 'tcp_error' => $tcpError !== '' ? substr($tcpError, 0, 100) : null, + 'raw_full' => $rawFull !== '' ? $rawFull : null, + 'connect_ms' => max(0, $connectMs), + 'read_ms' => max(0, $readMs), + 'tcp_ms' => max(0, $connectMs + $readMs), + ]); +} + +function tcp_queue_maybe_store_status(PDO $pdo, array $job, string $rawFull): void +{ + $trimError = ''; + $rawTrim = make_trim($rawFull, $trimError); + if ($rawTrim === '') { + return; + } + + $data = parse_trim($rawTrim); + if (!is_valid_status_data($data)) { + return; + } + + db_insert_status($pdo, (string)$job['cmd'], $rawFull, $rawTrim, $data); +} + +function tcp_worker_process_job(array $job): void +{ + $pdo = db(true); + db_ensure_tcp_queue_schema($pdo); + + $cmd = (string)$job['cmd']; + $requestId = (string)$job['request_id']; + $payload = (string)$job['request_payload']; + $uiDeadline = (float)$job['ui_deadline_ts']; + $usageDeadline = (float)$job['usage_deadline_ts']; + $connectStart = microtime(true); + $sentBytes = 0; + $receivedBytes = 0; + $response = ''; + $tcpError = ''; + $uiTimedOut = false; + + if (microtime(true) >= $uiDeadline) { + tcp_queue_record_usage_and_status($pdo, $job, 'expired_before_send', 0, 0, false, 'queue_expired_before_send', '', 0, 0); + return; + } + + $connectTimeout = max(0.01, min(TCP_TOTAL_TIMEOUT, $uiDeadline - microtime(true))); + $fp = @stream_socket_client("tcp://" . TCP_HOST . ":" . TCP_PORT, $errno, $errstr, $connectTimeout); + $connectMs = (int)round((microtime(true) - $connectStart) * 1000); + + if (!$fp) { + tcp_queue_record_usage_and_status($pdo, $job, 'connect_error', 0, 0, false, "connect_error: $errno $errstr", '', $connectMs, 0); + return; + } + + stream_set_blocking($fp, false); + $written = @fwrite($fp, $payload); + if ($written === false || $written < strlen($payload)) { + $sentBytes = ($written === false) ? 0 : (int)$written; + @fclose($fp); + tcp_queue_record_usage_and_status($pdo, $job, 'write_failed', $sentBytes, 0, false, 'write_failed', '', $connectMs, 0); + return; + } + + $sentBytes = (int)$written; + tcp_queue_update($pdo, $requestId, [ + 'sent_bytes' => $sentBytes, + 'total_bytes' => $sentBytes, + ]); + + $readStart = microtime(true); + while (microtime(true) < $usageDeadline) { + $now = microtime(true); + + if (!$uiTimedOut && $now >= $uiDeadline) { + $uiTimedOut = true; + tcp_queue_update($pdo, $requestId, [ + 'status' => 'ui_timeout_pending', + 'tcp_error' => 'total_timeout_4_95s_late_settlement_pending', + 'received_bytes' => strlen($response), + 'total_bytes' => $sentBytes + strlen($response), + 'connect_ms' => $connectMs, + 'read_ms' => (int)round(($now - $readStart) * 1000), + 'tcp_ms' => $connectMs + (int)round(($now - $readStart) * 1000), + ]); + } + + $nextWake = min($usageDeadline, $uiTimedOut ? $now + 1.0 : $uiDeadline); + $wait = max(0.001, min(1.0, $nextWake - $now)); $sec = (int)$wait; $usec = (int)(($wait - $sec) * 1000000); + $read = [$fp]; + $w = null; + $e = null; $sel = @stream_select($read, $w, $e, $sec, $usec); if ($sel === false) { - $tcpError = 'late_stream_select_failed'; + $tcpError = 'stream_select_failed'; break; } if ($sel > 0) { $chunk = @fread($fp, 4096); if ($chunk === false) { - $tcpError = 'late_read_failed'; + $tcpError = 'read_failed'; break; } if ($chunk !== '') { $response .= $chunk; + $receivedBytes = strlen($response); + tcp_queue_update($pdo, $requestId, [ + 'received_bytes' => $receivedBytes, + 'total_bytes' => $sentBytes + $receivedBytes, + ]); if (preg_match(RAW_FULL_REGEX, trim($response))) { break; @@ -216,77 +466,87 @@ function finalize_late_tcp_usage( @fclose($fp); + $readMs = (int)round((microtime(true) - $readStart) * 1000); $receivedBytes = strlen($response); $trimmed = trim($response); if ($trimmed === '') { - db_insert_data_usage(db(true), $usageSource, $cmd, $sentBytes, 0, false, 'late_total_timeout_60s_sent_only', $requestId); + tcp_queue_record_usage_and_status($pdo, $job, 'settled_sent_only', $sentBytes, 0, false, 'late_total_timeout_60s_sent_only', '', $connectMs, $readMs); return; } if (!preg_match(RAW_FULL_REGEX, $trimmed)) { - db_insert_data_usage( - db(true), - $usageSource, - $cmd, + tcp_queue_record_usage_and_status( + $pdo, + $job, + 'settled_partial', $sentBytes, $receivedBytes, false, $tcpError !== '' ? $tcpError : 'late_invalid_or_partial_response', - $requestId + $trimmed, + $connectMs, + $readMs ); return; } - db_insert_data_usage(db(true), $usageSource, $cmd, $sentBytes, $receivedBytes, true, 'late_response_after_ui_timeout', $requestId); - - $trimError = ''; - $rawTrim = make_trim($trimmed, $trimError); - if ($rawTrim === '') { - return; - } - - $data = parse_trim($rawTrim); - if (!is_valid_status_data($data)) { - return; - } - - try { - db_insert_status(db(true), $cmd, $trimmed, $rawTrim, $data); - } catch (Throwable $e) { - // 늦은 응답 상태 저장 실패가 사용량 정산을 되돌리면 안 된다. - } + $status = $uiTimedOut ? 'settled_late' : 'settled'; + $error = $uiTimedOut ? 'late_response_after_ui_timeout' : ''; + tcp_queue_record_usage_and_status($pdo, $job, $status, $sentBytes, $receivedBytes, true, $error, $trimmed, $connectMs, $readMs); + tcp_queue_maybe_store_status($pdo, $job, $trimmed); } -function schedule_late_tcp_usage_settlement( - $fp, - string $cmd, - string $usageSource, - ?string $requestId, - int $sentBytes, - string $initialResponse = '' -): bool { - if (!function_exists('pcntl_fork') || $sentBytes <= 0) { - return false; +function tcp_request_wait_for_queue( + string $requestId, + int &$tcpMs, + int &$connectMs, + int &$readMs, + string &$tcpError, + int &$sentBytes, + int &$receivedBytes +): string { + $deadline = microtime(true) + TCP_TOTAL_TIMEOUT; + + while (microtime(true) < $deadline) { + $row = tcp_queue_get($requestId); + if (!$row) { + usleep(50000); + continue; + } + + $status = (string)$row['status']; + $sentBytes = (int)($row['sent_bytes'] ?? 0); + $receivedBytes = (int)($row['received_bytes'] ?? 0); + $connectMs = (int)($row['connect_ms'] ?? 0); + $readMs = (int)($row['read_ms'] ?? 0); + $tcpMs = (int)($row['tcp_ms'] ?? ($connectMs + $readMs)); + $tcpError = (string)($row['tcp_error'] ?? ''); + + if ($status === 'settled' && (int)($row['tcp_ok'] ?? 0) === 1) { + return (string)($row['raw_full'] ?? ''); + } + + if (in_array($status, ['ui_timeout_pending', 'connect_error', 'write_failed', 'settled_sent_only', 'settled_partial'], true)) { + if ($tcpError === '') { + $tcpError = $status; + } + return ''; + } + + usleep(50000); } - $deadline = microtime(true) + TCP_USAGE_SETTLEMENT_TIMEOUT; - $pid = @pcntl_fork(); - - if ($pid === -1) { - return false; + $row = tcp_queue_get($requestId); + if ($row) { + $sentBytes = (int)($row['sent_bytes'] ?? 0); + $receivedBytes = (int)($row['received_bytes'] ?? 0); + $connectMs = (int)($row['connect_ms'] ?? 0); + $readMs = (int)($row['read_ms'] ?? 0); + $tcpMs = (int)($row['tcp_ms'] ?? ($connectMs + $readMs)); } - - if ($pid > 0) { - return true; - } - - if (function_exists('posix_setsid')) { - @posix_setsid(); - } - - finalize_late_tcp_usage($fp, $cmd, $usageSource, $requestId, $sentBytes, $initialResponse, $deadline); - exit(0); + $tcpError = 'queue_wait_timeout_late_settlement_pending'; + return ''; } function tcp_request( @@ -308,133 +568,27 @@ function tcp_request( $sentBytes = 0; $receivedBytes = 0; - $request = json_encode([ - 'type' => 'R', - 'type_sub' => 'car_controll', - 'data' => [ - 'command' => '+SCMD=' . MODEM . '/C:' . $cmd, - 'modem' => MODEM, - 'user' => USER, - 'uid' => UID, - 'type' => TYPE, - ], - ], JSON_UNESCAPED_SLASHES); - - $t0 = microtime(true); - - $deadline = $t0 + TCP_TOTAL_TIMEOUT; - - $remain = $deadline - microtime(true); - if ($remain <= 0) { - record_tcp_usage($usageSource, $cmd, 0, 0, false, 'deadline_expired_before_connect', $requestId); - return ''; + if ($requestId === null || $requestId === '') { + $requestId = substr(bin2hex(random_bytes(16)), 0, 32); } - - $fp = @stream_socket_client( - "tcp://" . TCP_HOST . ":" . TCP_PORT, - $errno, - $errstr, - $remain - ); - $connectMs = (int)round((microtime(true) - $t0) * 1000); - - if (!$fp) { - $tcpError = "connect_error: $errno $errstr"; - $tcpMs = $connectMs; + + try { + tcp_queue_insert($cmd, $usageSource, $requestId); + } catch (Throwable $e) { + $tcpError = 'queue_insert_failed'; record_tcp_usage($usageSource, $cmd, 0, 0, false, $tcpError, $requestId); return ''; } - stream_set_blocking($fp, false); - - $written = @fwrite($fp, $request); - if ($written === false || $written < strlen($request)) { - $sentBytes = ($written === false) ? 0 : (int)$written; - fclose($fp); - $tcpError = 'write_failed'; - record_tcp_usage($usageSource, $cmd, $sentBytes, 0, false, $tcpError, $requestId); - return ''; - } - $sentBytes = (int)$written; - - $r0 = microtime(true); - $response = ''; - while (microtime(true) < $deadline) { - $read = [$fp]; - $w = null; - $e = null; - - $remain = $deadline - microtime(true); - if ($remain <= 0) break; - - $sec = (int)$remain; - $usec = (int)(($remain - $sec) * 1000000); - - $sel = @stream_select($read, $w, $e, $sec, $usec); - if ($sel === false) { - $tcpError = 'stream_select_failed'; - break; - } - - if ($sel > 0) { - $chunk = @fread($fp, 4096); - if ($chunk === false) { - $tcpError = 'read_failed'; - break; - } - - if ($chunk !== '') { - $response .= $chunk; - - if (preg_match(RAW_FULL_REGEX, trim($response))) { - break; - } - } - } - - if (feof($fp)) { - break; - } - } - - $readMs = (int)round((microtime(true) - $r0) * 1000); - - $tcpMs = $connectMs + $readMs; - $receivedBytes = strlen($response); - - if (trim($response) === '') { - if ($tcpError === '') { - $tcpError = 'total_timeout_4_95s'; - } - $scheduled = $tcpError === 'total_timeout_4_95s' - && schedule_late_tcp_usage_settlement($fp, $cmd, $usageSource, $requestId, $sentBytes, $response); - if ($scheduled) { - fclose($fp); - $tcpError = 'total_timeout_4_95s_late_settlement_pending'; - } else { - fclose($fp); - record_tcp_usage($usageSource, $cmd, $sentBytes, $receivedBytes, false, $tcpError, $requestId); - } - return ''; - } - - if (!preg_match(RAW_FULL_REGEX, trim($response))) { - $tcpError = 'invalid_or_partial_response'; - $scheduled = schedule_late_tcp_usage_settlement($fp, $cmd, $usageSource, $requestId, $sentBytes, $response); - if ($scheduled) { - fclose($fp); - $tcpError = 'invalid_or_partial_response_late_settlement_pending'; - } else { - fclose($fp); - record_tcp_usage($usageSource, $cmd, $sentBytes, $receivedBytes, false, $tcpError, $requestId); - } - return ''; - } - - fclose($fp); - record_tcp_usage($usageSource, $cmd, $sentBytes, $receivedBytes, true, $tcpError, $requestId); - - return $response; + return tcp_request_wait_for_queue( + $requestId, + $tcpMs, + $connectMs, + $readMs, + $tcpError, + $sentBytes, + $receivedBytes + ); } function make_trim(string $rawFull, string &$trimError = ''): string diff --git a/tcp_worker.php b/tcp_worker.php new file mode 100644 index 0000000..a66dfb6 --- /dev/null +++ b/tcp_worker.php @@ -0,0 +1,44 @@ +getMessage() . "\n"); + usleep(1000000); + } +}