Fixes
This commit is contained in:
parent
5da683f508
commit
b4114cd5f8
@ -106,6 +106,12 @@ class FeedLoop extends ResumableSignalLoop
|
||||
$key = key($updates);
|
||||
$update = $updates[$key];
|
||||
unset($updates[$key]);
|
||||
if ($update['_'] === 'updateChannelTooLong') {
|
||||
$this->API->logger->logger('Got channel too long update, getting difference...', \danog\MadelineProto\Logger::VERBOSE);
|
||||
$this->API->updaters[$this->channelId]->resumeDefer();
|
||||
|
||||
continue;
|
||||
}
|
||||
if (isset($update['pts'])) {
|
||||
$logger = function ($msg) use ($update) {
|
||||
$pts_count = isset($update['pts_count']) ? $update['pts_count'] : 0;
|
||||
@ -123,7 +129,7 @@ class FeedLoop extends ResumableSignalLoop
|
||||
}
|
||||
if ($result > 0) {
|
||||
$logger("PTS hole");
|
||||
$this->updater->setLimit($state->pts + $result);
|
||||
$this->updater->setLimit($this->state->pts() + $result);
|
||||
yield $this->updater->resume();
|
||||
$updates = array_merge($this->incomingUpdates, $updates);
|
||||
$this->incomingUpdates = null;
|
||||
|
@ -76,7 +76,7 @@ class SeqLoop extends ResumableSignalLoop
|
||||
|
||||
return;
|
||||
}
|
||||
if (!$this->settings['updates']['handle_updates']) {
|
||||
if (!$this->API->settings['updates']['handle_updates']) {
|
||||
$API->logger->logger("Exiting update seq loop");
|
||||
$this->exitedLoop();
|
||||
return;
|
||||
@ -85,7 +85,7 @@ class SeqLoop extends ResumableSignalLoop
|
||||
while ($this->incomingUpdates) {
|
||||
$updates = $this->incomingUpdates;
|
||||
$this->incomingUpdates = [];
|
||||
$result += yield $this->parse($updates);
|
||||
$result = array_merge(yield $this->parse($updates), $result);
|
||||
$updates = null;
|
||||
}
|
||||
foreach ($result as $channelId => $boh) {
|
||||
@ -95,22 +95,22 @@ class SeqLoop extends ResumableSignalLoop
|
||||
}
|
||||
public function parse($updates)
|
||||
{
|
||||
$result = [];
|
||||
$fresult = [];
|
||||
reset($updates);
|
||||
while ($updates) {
|
||||
$options = [];
|
||||
$key = key($updates);
|
||||
$update = $updates[$key];
|
||||
unset($updates[$key]);
|
||||
|
||||
|
||||
$options = $update['options'];
|
||||
$updates = $update['updates'];
|
||||
|
||||
$seq_start = $options['seq_start'];
|
||||
$seq_end = $options['seq_end'];
|
||||
|
||||
$result = $this->state->checkSeq($seq_start);
|
||||
if ($result > 0) {
|
||||
$this->logger->logger('Seq hole. seq_start: '.$seq_start.' != cur seq: '.($this->state->seq() + 1), \danog\MadelineProto\Logger::ERROR);
|
||||
$this->API->logger->logger('Seq hole. seq_start: '.$seq_start.' != cur seq: '.($this->state->seq() + 1), \danog\MadelineProto\Logger::ERROR);
|
||||
yield $this->pause(1.0);
|
||||
if (!$this->incomingUpdates) {
|
||||
yield $this->updaters[false]->resume();
|
||||
@ -120,7 +120,7 @@ class SeqLoop extends ResumableSignalLoop
|
||||
continue;
|
||||
}
|
||||
if ($result < 0) {
|
||||
$this->logger->logger('Seq too old. seq_start: '.$seq_start.' != cur seq: '.($this->state->seq() + 1), \danog\MadelineProto\Logger::ERROR);
|
||||
$this->API->logger->logger('Seq too old. seq_start: '.$seq_start.' != cur seq: '.($this->state->seq() + 1), \danog\MadelineProto\Logger::ERROR);
|
||||
continue;
|
||||
}
|
||||
$this->state->seq($seq_end);
|
||||
@ -128,9 +128,9 @@ class SeqLoop extends ResumableSignalLoop
|
||||
$this->state->date($options['date']);
|
||||
}
|
||||
|
||||
$result += $this->save($updates);
|
||||
$fresult = array_merge(yield $this->save($update), $fresult);
|
||||
}
|
||||
return $result;
|
||||
return $fresult;
|
||||
}
|
||||
public function feed($updates)
|
||||
{
|
||||
|
@ -80,7 +80,7 @@ class UpdateLoop extends ResumableSignalLoop
|
||||
} else {
|
||||
$limit = 100;
|
||||
}
|
||||
$difference = yield $this->method_call_async_read('updates.getChannelDifference', ['channel' => 'channel#'.$this->channelId, 'filter' => ['_' => 'channelMessagesFilterEmpty'], 'pts' => $state->pts(), 'limit' => $limit, 'force' => true], ['datacenter' => $this->datacenter->curdc]);
|
||||
$difference = yield $this->API->method_call_async_read('updates.getChannelDifference', ['channel' => 'channel#'.$this->channelId, 'filter' => ['_' => 'channelMessagesFilterEmpty'], 'pts' => $state->pts(), 'limit' => $limit, 'force' => true], ['datacenter' => $this->API->datacenter->curdc]);
|
||||
if (isset($difference['timeout'])) {
|
||||
$timeout = $difference['timeout'];
|
||||
}
|
||||
|
@ -587,9 +587,9 @@ trait ResponseHandler
|
||||
|
||||
if ($updates['updates']) {
|
||||
if ($updates['_'] === 'updatesCombined') {
|
||||
$updates['updates'][0]['options'] = ['seq_start' => $updates['seq_start'], 'seq_end' => $updates['seq'], 'date' => $updates['date']];
|
||||
$updates['options'] = ['seq_start' => $updates['seq_start'], 'seq_end' => $updates['seq'], 'date' => $updates['date']];
|
||||
} else {
|
||||
$updates['updates'][0]['options'] = ['seq_start' => $updates['seq'], 'seq_end' => $updates['seq'], 'date' => $updates['date']];
|
||||
$updates['options'] = ['seq_start' => $updates['seq'], 'seq_end' => $updates['seq'], 'date' => $updates['date']];
|
||||
}
|
||||
$this->seqUpdater->feed($updates);
|
||||
$this->seqUpdater->resumeDefer();
|
||||
|
@ -23,6 +23,8 @@ use Amp\Artax\Request;
|
||||
use Amp\Deferred;
|
||||
use Amp\Delayed;
|
||||
use function Amp\Promise\any;
|
||||
use danog\MadelineProto\Loop\Update\FeedLoop;
|
||||
use danog\MadelineProto\Loop\Update\UpdateLoop;
|
||||
|
||||
/**
|
||||
* Manages updates.
|
||||
@ -180,11 +182,6 @@ trait UpdateHandler
|
||||
}
|
||||
|
||||
switch ($update['_']) {
|
||||
case 'updateChannelTooLong':
|
||||
$this->logger->logger('Got channel too long update, getting difference...', \danog\MadelineProto\Logger::VERBOSE);
|
||||
$this->updaters[$channelId]->resumeDefer();
|
||||
|
||||
return;
|
||||
case 'updateNewMessage':
|
||||
case 'updateEditMessage':
|
||||
case 'updateNewChannelMessage':
|
||||
|
@ -201,6 +201,6 @@ class UpdatesState
|
||||
*/
|
||||
public function checkSeq($seq)
|
||||
{
|
||||
return $seq - ($this->seq + 1);
|
||||
return $seq ? $seq - ($this->seq + 1) : $seq;
|
||||
}
|
||||
}
|
||||
|
Loading…
Reference in New Issue
Block a user