1
0
mirror of https://github.com/danog/MadelineProto.git synced 2024-12-02 17:17:48 +01:00
MadelineProto/src/VoIPController.php

741 lines
27 KiB
PHP
Raw Normal View History

2023-08-13 11:40:35 +02:00
<?php declare(strict_types=1);
/**
* This file is part of MadelineProto.
* MadelineProto is free software: you can redistribute it and/or modify it under the terms of the GNU Affero General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.
* MadelineProto is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
* See the GNU Affero General Public License for more details.
* You should have received a copy of the GNU General Public License along with MadelineProto.
* If not, see <http://www.gnu.org/licenses/>.
*
* @author Daniil Gentili <daniil@daniil.it>
* @copyright 2016-2023 Daniil Gentili <daniil@daniil.it>
* @license https://opensource.org/licenses/AGPL-3.0 AGPLv3
* @link https://docs.madelineproto.xyz MadelineProto documentation
*/
// Please keep the above notice the next time you copy my code, or I will sue you :)
2023-08-12 17:29:00 +02:00
namespace danog\MadelineProto;
2023-08-14 18:28:44 +02:00
use Amp\ByteStream\Pipe;
2023-08-13 18:31:31 +02:00
use Amp\ByteStream\ReadableStream;
2023-08-14 19:12:59 +02:00
use Amp\Cancellation;
use Amp\CancelledException;
use Amp\DeferredCancellation;
2023-08-14 20:17:19 +02:00
use Amp\DeferredFuture;
2023-08-14 19:12:59 +02:00
use danog\Loop\GenericLoop;
2023-08-12 17:29:00 +02:00
use danog\MadelineProto\MTProtoTools\Crypt;
use danog\MadelineProto\VoIP\CallState;
2023-08-12 21:47:09 +02:00
use danog\MadelineProto\VoIP\DiscardReason;
2023-08-12 17:29:00 +02:00
use danog\MadelineProto\VoIP\Endpoint;
use danog\MadelineProto\VoIP\MessageHandler;
use danog\MadelineProto\VoIP\VoIPState;
use phpseclib3\Math\BigInteger;
use Revolt\EventLoop;
2023-08-14 20:17:19 +02:00
use SplQueue;
2023-08-12 17:29:00 +02:00
use Throwable;
use Webmozart\Assert\Assert;
use function Amp\delay;
/** @internal */
final class VoIPController
{
const NET_TYPE_UNKNOWN = 0;
const NET_TYPE_GPRS = 1;
const NET_TYPE_EDGE = 2;
const NET_TYPE_3G = 3;
const NET_TYPE_HSPA = 4;
const NET_TYPE_LTE = 5;
const NET_TYPE_WIFI = 6;
const NET_TYPE_ETHERNET = 7;
const NET_TYPE_OTHER_HIGH_SPEED = 8;
const NET_TYPE_OTHER_LOW_SPEED = 9;
const NET_TYPE_DIALUP = 10;
const NET_TYPE_OTHER_MOBILE = 11;
const DATA_SAVING_NEVER = 0;
const DATA_SAVING_MOBILE = 1;
const DATA_SAVING_ALWAYS = 2;
const PROXY_NONE = 0;
const PROXY_SOCKS5 = 1;
const AUDIO_STATE_NONE = -1;
const AUDIO_STATE_CREATED = 0;
const AUDIO_STATE_CONFIGURED = 1;
const AUDIO_STATE_RUNNING = 2;
const PKT_INIT = 1;
const PKT_INIT_ACK = 2;
const PKT_STREAM_STATE = 3;
const PKT_STREAM_DATA = 4;
const PKT_UPDATE_STREAMS = 5;
const PKT_PING = 6;
const PKT_PONG = 7;
const PKT_STREAM_DATA_X2 = 8;
const PKT_STREAM_DATA_X3 = 9;
const PKT_LAN_ENDPOINT = 10;
const PKT_NETWORK_CHANGED = 11;
const PKT_SWITCH_PREF_RELAY = 12;
const PKT_SWITCH_TO_P2P = 13;
const PKT_NOP = 14;
const TLID_DECRYPTED_AUDIO_BLOCK = "\xc1\xdb\xf9\x48";
const TLID_SIMPLE_AUDIO_BLOCK = "\x0d\x0e\x76\xcc";
const TLID_REFLECTOR_SELF_INFO = "\xC7\x72\x15\xc0";
const TLID_REFLECTOR_PEER_INFO = "\x1C\x37\xD9\x27";
const PROTO_ID = 'GrVP';
const PROTOCOL_VERSION = 9;
const MIN_PROTOCOL_VERSION = 9;
const STREAM_TYPE_AUDIO = 1;
const STREAM_TYPE_VIDEO = 2;
const CODEC_OPUS = 'SUPO';
private MessageHandler $messageHandler;
private VoIPState $voipState = VoIPState::CREATED;
private CallState $callState;
private array $call;
2023-08-14 17:16:59 +02:00
/** @var array<LocalFile|RemoteUrl|ReadableStream> */
2023-08-12 17:29:00 +02:00
private array $holdFiles = [];
2023-08-14 17:16:59 +02:00
/** @var list<LocalFile|RemoteUrl|ReadableStream> */
2023-08-12 17:29:00 +02:00
private array $inputFiles = [];
2023-08-12 21:47:09 +02:00
private int $holdIndex = 0;
2023-08-12 17:29:00 +02:00
/**
* @var array<Endpoint>
*/
private array $sockets = [];
2023-08-12 21:47:09 +02:00
private Endpoint $bestEndpoint;
2023-08-12 22:13:23 +02:00
private ?string $pendingPing = null;
2023-08-12 17:29:00 +02:00
private ?string $timeoutWatcher = null;
private float $lastIncomingTimestamp = 0.0;
private float $lastOutgoingTimestamp = 0.0;
2023-08-12 17:29:00 +02:00
private int $opusTimestamp = 0;
2023-08-14 20:17:19 +02:00
private SplQueue $packetQueue;
private ?DeferredFuture $packetDeferred = null;
2023-08-14 19:12:59 +02:00
private Cancellation $cancellation;
private DeferredCancellation $deferred;
private GenericLoop $playLoop;
2023-08-12 17:29:00 +02:00
/** Auth key */
private readonly string $authKey;
public readonly VoIP $public;
/** @var ?list{string, string, string, string} */
private ?array $visualization = null;
/**
* Constructor.
*
* @internal
*/
public function __construct(
private readonly MTProto $API,
array $call
) {
$this->public = new VoIP($API, $call);
$call['_'] = 'inputPhoneCall';
2023-08-14 20:17:19 +02:00
$this->packetQueue = new SplQueue();
2023-08-14 19:12:59 +02:00
$this->deferred = new DeferredCancellation;
$this->cancellation = $this->deferred->getCancellation();
$this->playLoop = new GenericLoop($this->playLoop(...), "Play loop");
2023-08-14 20:17:19 +02:00
Assert::true($this->playLoop->start());
2023-08-12 17:29:00 +02:00
$this->call = $call;
2023-08-12 21:47:09 +02:00
if ($this->public->outgoing) {
2023-08-12 17:29:00 +02:00
$this->callState = CallState::REQUESTED;
2023-08-12 21:47:09 +02:00
} else {
$this->callState = CallState::INCOMING;
2023-08-12 17:29:00 +02:00
}
}
2023-08-14 20:17:19 +02:00
public function __serialize(): array
{
$data = \get_object_vars($this);
unset($data['cancellation'], $data['deferred'], $data['packetDeferred'], $data['playLoop']);
$data['holdFiles'] = \array_filter(
$data['holdFiles'],
fn ($v) => !$v instanceof ReadableStream
);
$data['inputFiles'] = \array_filter(
$data['inputFiles'],
fn ($v) => !$v instanceof ReadableStream
);
return $data;
}
/**
* Wakeup function.
*/
public function __unserialize(array $data): void
{
foreach ($data as $key => $value) {
$this->{$key} = $value;
}
$this->deferred = new DeferredCancellation;
$this->cancellation = $this->deferred->getCancellation();
$this->playLoop = new GenericLoop($this->playLoop(...), "Play loop");
Assert::true($this->playLoop->start());
if ($this->callState === CallState::RUNNING) {
$this->lastIncomingTimestamp = \microtime(true);
$this->startReadLoop();
if ($this->pendingPing) {
$this->pendingPing = EventLoop::repeat(0.2, $this->ping(...));
}
if ($this->voipState === VoIPState::ESTABLISHED) {
$diff = (int) ((\microtime(true) - $this->lastOutgoingTimestamp) * 1000);
$this->opusTimestamp += $diff - ($diff % 60);
EventLoop::queue($this->startWriteLoop(...));
}
}
}
2023-08-12 17:29:00 +02:00
/**
* Confirm requested call.
* @internal
*/
public function confirm(array $params): bool
{
if ($this->callState !== CallState::REQUESTED) {
$this->API->logger->logger(\sprintf(Lang::$current_lang['call_error_2'], $this->public->callID));
return false;
}
$this->API->logger->logger(\sprintf(Lang::$current_lang['call_confirming'], $this->public->otherID), Logger::VERBOSE);
$dh_config = $this->API->getDhConfig();
$params['g_b'] = new BigInteger((string) $params['g_b'], 256);
Crypt::checkG($params['g_b'], $dh_config['p']);
$key = \str_pad($params['g_b']->powMod($this->call['a'], $dh_config['p'])->toBytes(), 256, \chr(0), STR_PAD_LEFT);
try {
$res = ($this->API->methodCallAsyncRead('phone.confirmCall', ['key_fingerprint' => \substr(\sha1($key, true), -8), 'peer' => ['id' => $params['id'], 'access_hash' => $params['access_hash'], '_' => 'inputPhoneCall'], 'g_a' => $this->call['g_a'], 'protocol' => ['_' => 'phoneCallProtocol', 'udp_reflector' => true, 'min_layer' => 65, 'max_layer' => 92]]))['phone_call'];
} catch (RPCErrorException $e) {
if ($e->rpc === 'CALL_ALREADY_ACCEPTED') {
$this->API->logger->logger(\sprintf(Lang::$current_lang['call_already_accepted'], $params['id']));
return true;
}
if ($e->rpc === 'CALL_ALREADY_DECLINED') {
$this->API->logger->logger(Lang::$current_lang['call_already_declined']);
2023-08-12 21:47:09 +02:00
$this->discard(DiscardReason::HANGUP);
2023-08-12 17:29:00 +02:00
return false;
}
throw $e;
}
$visualization = [];
$length = new BigInteger(\count(Magic::$emojis));
foreach (\str_split(\hash('sha256', $key.\str_pad($this->call['g_a'], 256, \chr(0), STR_PAD_LEFT), true), 8) as $number) {
$number[0] = \chr(\ord($number[0]) & 0x7f);
$visualization[] = Magic::$emojis[(int) (new BigInteger($number, 256))->divide($length)[1]->toString()];
}
$this->visualization = $visualization;
$this->authKey = $key;
$this->callState = CallState::RUNNING;
$this->messageHandler = new MessageHandler(
$this,
\substr(\hash('sha256', $key, true), -16)
);
$this->initialize($res['connections']);
return true;
}
/**
* Accept incoming call.
*/
public function accept(): self
{
if ($this->callState === CallState::RUNNING || $this->callState === CallState::ENDED) {
return $this;
}
Assert::eq($this->callState->name, CallState::INCOMING->name);
$this->API->logger->logger(\sprintf(Lang::$current_lang['accepting_call'], $this->public->otherID), Logger::VERBOSE);
$dh_config = $this->API->getDhConfig();
$this->API->logger->logger('Generating b...', Logger::VERBOSE);
$b = BigInteger::randomRange(Magic::$two, $dh_config['p']->subtract(Magic::$two));
$g_b = $dh_config['g']->powMod($b, $dh_config['p']);
Crypt::checkG($g_b, $dh_config['p']);
2023-08-12 21:47:09 +02:00
$this->callState = CallState::ACCEPTED;
2023-08-12 17:29:00 +02:00
try {
$this->API->methodCallAsyncRead('phone.acceptCall', ['peer' => ['id' => $this->call['id'], 'access_hash' => $this->call['access_hash'], '_' => 'inputPhoneCall'], 'g_b' => $g_b->toBytes(), 'protocol' => ['_' => 'phoneCallProtocol', 'udp_reflector' => true, 'udp_p2p' => true, 'min_layer' => 65, 'max_layer' => 92]]);
} catch (RPCErrorException $e) {
if ($e->rpc === 'CALL_ALREADY_ACCEPTED') {
$this->API->logger->logger(\sprintf(Lang::$current_lang['call_already_accepted'], $this->public->callID));
return $this;
}
if ($e->rpc === 'CALL_ALREADY_DECLINED') {
$this->API->logger->logger(Lang::$current_lang['call_already_declined']);
2023-08-12 21:47:09 +02:00
$this->discard(DiscardReason::HANGUP);
2023-08-12 17:29:00 +02:00
return $this;
}
throw $e;
}
$this->call['b'] = $b;
return $this;
}
/**
* Complete call handshake.
*
* @internal
*/
public function complete(array $params): bool
{
if ($this->callState !== CallState::ACCEPTED) {
$this->API->logger->logger(\sprintf(Lang::$current_lang['call_error_3'], $params['id']));
return false;
}
$this->API->logger->logger(\sprintf(Lang::$current_lang['call_completing'], $this->public->otherID), Logger::VERBOSE);
$dh_config = $this->API->getDhConfig();
2023-08-12 21:47:09 +02:00
if (\hash('sha256', (string) $params['g_a_or_b'], true) !== (string) $this->call['g_a_hash']) {
2023-08-12 17:29:00 +02:00
throw new SecurityException('Invalid g_a!');
}
$params['g_a_or_b'] = new BigInteger((string) $params['g_a_or_b'], 256);
Crypt::checkG($params['g_a_or_b'], $dh_config['p']);
$key = \str_pad($params['g_a_or_b']->powMod($this->call['b'], $dh_config['p'])->toBytes(), 256, \chr(0), STR_PAD_LEFT);
if (\substr(\sha1($key, true), -8) != $params['key_fingerprint']) {
throw new SecurityException(Lang::$current_lang['fingerprint_invalid']);
}
$visualization = [];
$length = new BigInteger(\count(Magic::$emojis));
foreach (\str_split(\hash('sha256', $key.\str_pad($params['g_a_or_b']->toBytes(), 256, \chr(0), STR_PAD_LEFT), true), 8) as $number) {
$number[0] = \chr(\ord($number[0]) & 0x7f);
$visualization[] = Magic::$emojis[(int) (new BigInteger($number, 256))->divide($length)[1]->toString()];
}
$this->visualization = $visualization;
$this->authKey = $key;
$this->callState = CallState::RUNNING;
$this->messageHandler = new MessageHandler(
$this,
\substr(\hash('sha256', $key, true), -16)
);
$this->initialize($params['connections']);
return true;
}
/**
* Get call emojis (will return null if the call is not inited yet).
*
* @return ?list{string, string, string, string}
*/
public function getVisualization(): ?array
{
return $this->visualization;
}
/**
* Discard call.
2023-08-12 21:47:09 +02:00
*
* @param int<1, 5> $rating Call rating in stars
* @param string $comment Additional comment on call quality.
2023-08-12 17:29:00 +02:00
*/
2023-08-12 21:47:09 +02:00
public function discard(DiscardReason $reason = DiscardReason::HANGUP, ?int $rating = null, ?string $comment = null): self
2023-08-12 17:29:00 +02:00
{
if ($this->callState === CallState::ENDED) {
return $this;
}
2023-08-12 21:47:09 +02:00
$this->API->cleanupCall($this->public->callID);
2023-08-14 20:17:19 +02:00
$this->callState = CallState::ENDED;
$this->playLoop->stop();
$this->deferred->cancel();
$this->skip();
2023-08-12 21:47:09 +02:00
2023-08-12 17:29:00 +02:00
Logger::log("Now closing $this");
if (isset($this->timeoutWatcher)) {
EventLoop::cancel($this->timeoutWatcher);
}
2023-08-12 22:13:23 +02:00
if (isset($this->pendingPing)) {
EventLoop::cancel($this->pendingPing);
}
2023-08-12 17:29:00 +02:00
Logger::log("Closing all sockets in $this");
foreach ($this->sockets as $socket) {
$socket->disconnect();
}
2023-08-14 20:17:19 +02:00
$this->packetQueue = new SplQueue;
$this->packetDeferred?->complete(false);
2023-08-12 17:29:00 +02:00
Logger::log("Closed all sockets, discarding $this");
$this->API->logger->logger(\sprintf(Lang::$current_lang['call_discarding'], $this->public->callID), Logger::VERBOSE);
try {
2023-08-12 21:47:09 +02:00
$this->API->methodCallAsyncRead('phone.discardCall', ['peer' => $this->call, 'duration' => \time() - $this->public->date, 'connection_id' => 0, 'reason' => ['_' => match ($reason) {
DiscardReason::BUSY => 'phoneCallDiscardReasonBusy',
DiscardReason::HANGUP => 'phoneCallDiscardReasonHangup',
DiscardReason::DISCONNECTED => 'phoneCallDiscardReasonDisconnect',
DiscardReason::MISSED => 'phoneCallDiscardReasonMissed'
}]]);
2023-08-12 17:29:00 +02:00
} catch (RPCErrorException $e) {
if (!\in_array($e->rpc, ['CALL_ALREADY_DECLINED', 'CALL_ALREADY_ACCEPTED'], true)) {
throw $e;
}
}
2023-08-12 21:47:09 +02:00
if ($rating !== null) {
2023-08-12 17:29:00 +02:00
$this->API->logger->logger(\sprintf('Setting rating for call %s...', $this->call), Logger::VERBOSE);
2023-08-12 21:47:09 +02:00
$this->API->methodCallAsyncRead('phone.setCallRating', ['peer' => $this->call, 'rating' => $rating, 'comment' => $comment]);
2023-08-12 17:29:00 +02:00
}
return $this;
}
/**
* Connect to the specified endpoints.
*/
private function initialize(array $endpoints): void
{
foreach ($endpoints as $endpoint) {
try {
$this->sockets['v6 '.$endpoint['id']] = new Endpoint(
'['.$endpoint['ipv6'].']',
$endpoint['port'],
$endpoint['peer_tag'],
true,
$this->public->outgoing,
$this->authKey,
$this->messageHandler
);
} catch (Throwable) {
}
try {
$this->sockets['v4 '.$endpoint['id']] = new Endpoint(
$endpoint['ip'],
$endpoint['port'],
$endpoint['peer_tag'],
true,
$this->public->outgoing,
$this->authKey,
$this->messageHandler
);
} catch (Throwable) {
}
}
$this->voipState = VoIPState::WAIT_INIT;
$this->startReadLoop();
foreach ($this->sockets as $socket) {
$socket->udpPing();
$socket->write($this->messageHandler->encryptPacket([
'_' => self::PKT_INIT,
'protocol' => self::PROTOCOL_VERSION,
'min_protocol' => self::MIN_PROTOCOL_VERSION,
'audio_streams' => [self::CODEC_OPUS],
'video_streams' => []
], true));
}
}
/**
* Handle incoming packet.
*/
private function handlePacket(Endpoint $socket, array $packet): void
{
switch ($packet['_']) {
case self::PKT_INIT:
//$this->voipState = VoIPState::WAIT_INIT_ACK;
$socket->write($this->messageHandler->encryptPacket([
'_' => self::PKT_INIT_ACK,
'protocol' => self::PROTOCOL_VERSION,
'min_protocol' => self::MIN_PROTOCOL_VERSION,
'all_streams' => [
['id' => 0, 'type' => self::STREAM_TYPE_AUDIO, 'codec' => self::CODEC_OPUS, 'frame_duration' => 60, 'enabled' => 1]
]
]));
$socket->write($this->messageHandler->encryptPacket([
'_' => self::PKT_INIT,
'protocol' => self::PROTOCOL_VERSION,
'min_protocol' => self::MIN_PROTOCOL_VERSION,
'audio_streams' => [self::CODEC_OPUS],
'video_streams' => []
]));
break;
case self::PKT_INIT_ACK:
2023-08-12 21:47:09 +02:00
if (!isset($this->bestEndpoint)) {
2023-08-12 17:29:00 +02:00
$this->bestEndpoint = $socket;
2023-08-12 22:13:23 +02:00
$this->pendingPing = EventLoop::repeat(0.2, $this->ping(...));
2023-08-12 17:29:00 +02:00
}
break;
case self::PKT_STREAM_DATA:
$cnt = 1;
break;
case self::PKT_STREAM_DATA_X2:
$cnt = 2;
break;
case self::PKT_STREAM_DATA_X3:
$cnt = 3;
break;
case self::PKT_PING:
$socket->write($this->messageHandler->encryptPacket(['_' => self::PKT_PONG, 'out_seq_no' => $packet['out_seq_no']]));
break;
case self::PKT_PONG:
if ($this->pendingPing) {
2023-08-12 22:13:23 +02:00
EventLoop::cancel($this->pendingPing);
$this->pendingPing = null;
2023-08-12 17:29:00 +02:00
if ($this->bestEndpoint !== $socket) {
Logger::log("Changing best endpoint from {$this->bestEndpoint} to $socket");
$this->bestEndpoint = $socket;
}
2023-08-14 19:12:59 +02:00
$this->muted = true;
$this->bestEndpoint->writeReliably([
'_' => self::PKT_STREAM_STATE,
'id' => 0,
'enabled' => false
]);
2023-08-12 22:13:23 +02:00
$this->startWriteLoop();
2023-08-12 17:29:00 +02:00
}
break;
}
}
2023-08-12 22:13:23 +02:00
private function ping(): void
{
foreach ($this->sockets as $socket) {
EventLoop::queue(fn () => $socket->write($this->messageHandler->encryptPacket(['_' => self::PKT_PING])));
}
}
2023-08-12 17:29:00 +02:00
private function startReadLoop(): void
{
foreach ($this->sockets as $socket) {
EventLoop::queue(function () use ($socket): void {
2023-08-15 10:14:43 +02:00
while (true) {
try {
$payload = $socket->read();
} catch (Throwable $e) {
Logger::log("Got $e in this!");
continue;
}
if (!$payload) {
break;
}
2023-08-12 17:29:00 +02:00
$this->lastIncomingTimestamp = \microtime(true);
EventLoop::queue($this->handlePacket(...), $socket, $payload);
}
Logger::log("Exiting VoIP read loop in $this!");
});
}
}
2023-08-14 18:28:44 +02:00
private function startPlaying(LocalFile|RemoteUrl|ReadableStream $f): void
{
$it = null;
if ($f instanceof LocalFile) {
try {
2023-08-14 19:12:59 +02:00
$it = new Ogg($f, $this->cancellation);
2023-08-14 18:28:44 +02:00
if (!\in_array('MADELINE_ENCODER_V=1', $it->comments, true)) {
$it = null;
}
} catch (Throwable) {
$it = null;
}
}
if (!$it) {
EventLoop::queue(Logger::log(...), "Starting conversion fiber...");
$pipe = new Pipe(4096);
2023-08-14 19:12:59 +02:00
EventLoop::queue(Ogg::convert(...), $f, $pipe->getSink(), $this->cancellation);
2023-08-14 18:28:44 +02:00
$it = new Ogg($pipe->getSource());
}
foreach ($it->opusPackets as $packet) {
2023-08-14 20:17:19 +02:00
$this->packetQueue->enqueue($packet);
if ($this->packetDeferred) {
$deferred = $this->packetDeferred;
$this->packetDeferred = null;
$deferred->complete(true);
}
2023-08-14 18:28:44 +02:00
}
}
2023-08-12 21:47:09 +02:00
private bool $muted = false;
private bool $playingHold = false;
2023-08-14 20:17:19 +02:00
private function playLoop(): ?float
2023-08-12 21:47:09 +02:00
{
2023-08-14 20:17:19 +02:00
if ($this->callState === CallState::ENDED) {
Logger::log("Exiting play loop in $this because the call ended!");
return GenericLoop::STOP;
}
Logger::log("Starting play loop in $this!");
$file = \array_shift($this->inputFiles);
2023-08-14 20:54:16 +02:00
if (!$file) {
Logger::log("Pausing play loop in $this!");
return GenericLoop::PAUSE;
2023-08-14 20:17:19 +02:00
}
try {
$this->startPlaying($file);
} catch (CancelledException) {
$this->packetQueue = new SplQueue;
if ($this->packetDeferred) {
$deferred = $this->packetDeferred;
$this->packetDeferred = null;
$deferred?->complete(false);
2023-08-14 19:17:50 +02:00
}
}
2023-08-14 20:17:19 +02:00
return GenericLoop::CONTINUE;
2023-08-14 19:12:59 +02:00
}
private function pullPacket(): ?string
{
2023-08-14 20:17:19 +02:00
if ($this->packetQueue->isEmpty()) {
2023-08-14 20:54:16 +02:00
if ($this->callState === CallState::ENDED) {
return null;
}
if ($this->playLoop->isPaused()) {
if (!$this->holdFiles || $this->inputFiles) {
return null;
}
$this->playingHold = true;
$this->inputFiles []= $this->holdFiles[($this->holdIndex++) % \count($this->holdFiles)];
Assert::true($this->playLoop->resume());
2023-08-14 20:17:19 +02:00
return null;
}
$this->packetDeferred ??= new DeferredFuture;
if (!$this->packetDeferred->getFuture()->await()) {
return null;
}
2023-08-14 19:23:13 +02:00
}
2023-08-14 20:17:19 +02:00
return $this->packetQueue->dequeue();
2023-08-12 21:47:09 +02:00
}
2023-08-12 17:29:00 +02:00
/**
* Start write loop.
*/
private function startWriteLoop(): void
{
$this->voipState = VoIPState::ESTABLISHED;
2023-08-12 21:47:09 +02:00
Logger::log("Call established in $this, sending OPUS data!");
2023-08-12 17:29:00 +02:00
2023-08-12 22:13:23 +02:00
$this->timeoutWatcher = EventLoop::repeat(10, function (): void {
if (\microtime(true) - $this->lastIncomingTimestamp > 10) {
$this->discard(DiscardReason::DISCONNECTED);
}
});
2023-08-12 21:47:09 +02:00
$delay = $this->muted ? 0.2 : 0.06;
$t = \microtime(true) + $delay;
2023-08-12 17:29:00 +02:00
while (true) {
2023-08-14 17:16:59 +02:00
if ($packet = $this->pullPacket()) {
2023-08-12 21:47:09 +02:00
if ($this->muted) {
2023-08-14 17:16:59 +02:00
if (!$this->bestEndpoint->writeReliably([
2023-08-12 21:47:09 +02:00
'_' => self::PKT_STREAM_STATE,
'id' => 0,
'enabled' => true
2023-08-14 17:16:59 +02:00
])) {
2023-08-14 20:17:19 +02:00
Logger::log("Exiting write loop in $this because we could not write stream state!");
2023-08-14 17:16:59 +02:00
return;
}
2023-08-12 21:47:09 +02:00
$this->muted = false;
$delay = 0.06;
$this->opusTimestamp = 0;
2023-08-12 17:29:00 +02:00
}
2023-08-12 21:47:09 +02:00
$packet = $this->messageHandler->encryptPacket([
'_' => self::PKT_STREAM_DATA,
'stream_id' => 0,
2023-08-14 17:16:59 +02:00
'data' => $packet,
2023-08-12 21:47:09 +02:00
'timestamp' => $this->opusTimestamp
]);
$this->opusTimestamp += 60;
2023-08-14 17:16:59 +02:00
} else {
if (!$this->muted) {
if (!$this->bestEndpoint->writeReliably([
'_' => self::PKT_STREAM_STATE,
'id' => 0,
'enabled' => false
])) {
2023-08-14 20:17:19 +02:00
Logger::log("Exiting write loop in $this because we could not write stream state!");
2023-08-14 17:16:59 +02:00
return;
}
$this->muted = true;
$delay = 0.2;
}
$packet = $this->messageHandler->encryptPacket([
'_' => self::PKT_NOP
]);
2023-08-12 21:47:09 +02:00
}
//Logger::log("Writing {$this->opusTimestamp} in $this!");
$cur = \microtime(true);
$diff = $t - $cur;
if ($diff > 0) {
delay($diff);
2023-08-12 17:29:00 +02:00
}
2023-08-14 17:16:59 +02:00
if (!$this->bestEndpoint->write($packet)) {
2023-08-14 20:17:19 +02:00
Logger::log("Exiting write loop in $this!");
2023-08-14 17:16:59 +02:00
return;
}
2023-08-12 21:47:09 +02:00
if ($diff > 0) {
$cur += $diff;
2023-08-12 17:29:00 +02:00
}
2023-08-12 21:47:09 +02:00
$this->lastOutgoingTimestamp = $cur;
$t += $delay;
2023-08-12 17:29:00 +02:00
}
}
/**
* Play file.
*/
2023-08-14 16:10:20 +02:00
public function play(LocalFile|RemoteUrl|ReadableStream $file): self
2023-08-12 17:29:00 +02:00
{
$this->inputFiles[] = $file;
2023-08-12 21:47:09 +02:00
if ($this->playingHold) {
2023-08-14 20:54:16 +02:00
$this->playingHold = false;
2023-08-14 18:28:44 +02:00
$this->skip();
2023-08-12 21:47:09 +02:00
}
2023-08-15 12:41:45 +02:00
$this->playLoop->resume();
2023-08-12 17:29:00 +02:00
return $this;
}
2023-08-14 18:28:44 +02:00
/**
* When called, skips to the next file in the playlist.
*/
public function skip(): void
{
2023-08-14 20:17:19 +02:00
$this->packetQueue = new SplQueue;
2023-08-14 19:12:59 +02:00
$deferred = $this->deferred;
$this->deferred = new DeferredCancellation;
$this->cancellation = $this->deferred->getCancellation();
2023-08-14 19:17:50 +02:00
$deferred->cancel();
2023-08-14 18:28:44 +02:00
}
/**
* Stops playing all files, clears the main and the hold playlist.
*/
public function stop(): void
{
$this->inputFiles = [];
$this->holdFiles = [];
$this->skip();
}
2023-08-12 17:29:00 +02:00
/**
* Files to play on hold.
*/
2023-08-14 20:54:16 +02:00
public function playOnHold(LocalFile|RemoteUrl|ReadableStream ...$files): self
2023-08-12 17:29:00 +02:00
{
$this->holdFiles = $files;
return $this;
}
/**
* Get call state.
*/
public function getCallState(): CallState
{
return $this->callState;
}
/**
* Get VoIP state.
*/
public function getVoIPState(): VoIPState
{
return $this->voipState;
}
/**
* Get call representation.
*/
public function __toString(): string
{
return $this->public->__toString();
}
}