Avoid warnings in the Jetstream process

This commit is contained in:
Michael 2025-05-27 21:26:17 +00:00
parent dc0dee77d0
commit 2e701df21a
3 changed files with 45 additions and 6 deletions

View file

@ -13,6 +13,9 @@ if (php_sapi_name() !== 'cli') {
exit(); exit();
} }
// Ensure that te console is executed from the base path of the installation
chdir(dirname(__DIR__));
require dirname(__DIR__) . '/vendor/autoload.php'; require dirname(__DIR__) . '/vendor/autoload.php';
$container = \Friendica\Core\DiceContainer::fromBasePath(dirname(__DIR__)); $container = \Friendica\Core\DiceContainer::fromBasePath(dirname(__DIR__));

View file

@ -133,7 +133,7 @@ class Actor
} }
$directory = $this->atprotocol->get(ATProtocol::DIRECTORY . '/' . $profile->did); $directory = $this->atprotocol->get(ATProtocol::DIRECTORY . '/' . $profile->did);
if (!empty($directory)) { if (!empty($directory->service)) {
foreach ($directory->service as $service) { foreach ($directory->service as $service) {
if (($service->id == '#atproto_pds') && ($service->type == 'AtprotoPersonalDataServer') && !empty($service->serviceEndpoint)) { if (($service->id == '#atproto_pds') && ($service->type == 'AtprotoPersonalDataServer') && !empty($service->serviceEndpoint)) {
$fields['baseurl'] = $service->serviceEndpoint; $fields['baseurl'] = $service->serviceEndpoint;

View file

@ -84,6 +84,8 @@ class Jetstream
$timeout_limit = 10; $timeout_limit = 10;
$timestamp = $this->keyValue->get('jetstream_timestamp') ?? 0; $timestamp = $this->keyValue->get('jetstream_timestamp') ?? 0;
$cursor = ''; $cursor = '';
$this->logger->notice('Start listening');
while (true) { while (true) {
if ($timestamp) { if ($timestamp) {
$cursor = '&cursor=' . $timestamp; $cursor = '&cursor=' . $timestamp;
@ -92,12 +94,21 @@ class Jetstream
$this->syncContacts(); $this->syncContacts();
try { try {
set_error_handler(function ($errno, $errstr, $errfile, $errline) {
restore_error_handler();
throw new \ErrorException($errstr, 0, $errno, $errfile, $errline);
}, E_WARNING);
// @todo make the path configurable // @todo make the path configurable
$this->client = new \WebSocket\Client('wss://jetstream1.us-west.bsky.network/subscribe?requireHello=true' . $cursor); $this->client = new \WebSocket\Client('wss://jetstream1.us-west.bsky.network/subscribe?requireHello=true' . $cursor);
$this->client->setTimeout($timeout); $this->client->setTimeout($timeout);
$this->client->setLogger($this->logger); $this->client->setLogger($this->logger);
restore_error_handler();
} catch (\WebSocket\ConnectionException $e) { } catch (\WebSocket\ConnectionException $e) {
$this->logger->error('Error while trying to establish the connection', ['code' => $e->getCode(), 'message' => $e->getMessage()]); $this->logger->error('Error while trying to establish the connection', ['code' => $e->getCode(), 'message' => $e->getMessage(), 'file' => $e->getFile(), 'line' => $e->getLine()]);
echo "Connection wasn't established.\n";
exit(1);
} catch (\ErrorException $e) {
$this->logger->notice('Warning while trying to establish the connection', ['code' => $e->getCode(), 'severity' => $e->getSeverity(), 'message' => $e->getMessage(), 'file' => $e->getFile(), 'line' => $e->getLine()]);
echo "Connection wasn't established.\n"; echo "Connection wasn't established.\n";
exit(1); exit(1);
} }
@ -105,7 +116,13 @@ class Jetstream
$last_timeout = time(); $last_timeout = time();
while (true) { while (true) {
try { try {
$message = @$this->client->receive(); set_error_handler(function ($errno, $errstr, $errfile, $errline) {
restore_error_handler();
throw new \ErrorException($errstr, 0, $errno, $errfile, $errline);
}, E_WARNING);
$message = $this->client->receive();
restore_error_handler();
if (empty($message)) { if (empty($message)) {
$this->logger->notice('Empty message received'); $this->logger->notice('Empty message received');
break; break;
@ -128,19 +145,31 @@ class Jetstream
break; break;
} }
$this->logger->notice('Timeout', ['duration' => $timeout_duration, 'timestamp' => $timestamp, 'code' => $e->getCode(), 'message' => $e->getMessage()]); $this->logger->notice('Timeout', ['duration' => $timeout_duration, 'timestamp' => $timestamp, 'code' => $e->getCode(), 'message' => $e->getMessage()]);
break;
} else { } else {
$this->logger->error('Error', ['code' => $e->getCode(), 'message' => $e->getMessage()]); $this->logger->error('Error while trying to receive a message', ['code' => $e->getCode(), 'message' => $e->getMessage(), 'file' => $e->getFile(), 'line' => $e->getLine()]);
break; break;
} }
} catch (\ErrorException $e) {
$this->logger->notice('Warning while trying to receive a message', ['code' => $e->getCode(), 'severity' => $e->getSeverity(), 'message' => $e->getMessage(), 'file' => $e->getFile(), 'line' => $e->getLine()]);
break;
} }
$last_timeout = time(); $last_timeout = time();
} }
try { try {
set_error_handler(function ($errno, $errstr, $errfile, $errline) {
restore_error_handler();
throw new \ErrorException($errstr, 0, $errno, $errfile, $errline);
}, E_WARNING);
$this->client->close(); $this->client->close();
restore_error_handler();
} catch (\WebSocket\ConnectionException $e) { } catch (\WebSocket\ConnectionException $e) {
$this->logger->error('Error while trying to close the connection', ['code' => $e->getCode(), 'message' => $e->getMessage()]); $this->logger->error('Error while trying to close the connection', ['code' => $e->getCode(), 'message' => $e->getMessage(), 'file' => $e->getFile(), 'line' => $e->getLine()]);
} catch (\ErrorException $e) {
$this->logger->notice('Warning while trying to close the connection', ['code' => $e->getCode(), 'severity' => $e->getSeverity(), 'message' => $e->getMessage(), 'file' => $e->getFile(), 'line' => $e->getLine()]);
} }
} }
$this->logger->notice('Stop listening');
} }
/** /**
@ -234,9 +263,16 @@ class Jetstream
] ]
]; ];
try { try {
set_error_handler(function ($errno, $errstr, $errfile, $errline) {
restore_error_handler();
throw new \ErrorException($errstr, 0, $errno, $errfile, $errline);
}, E_WARNING);
$this->client->send(json_encode($update)); $this->client->send(json_encode($update));
restore_error_handler();
} catch (\WebSocket\ConnectionException $e) { } catch (\WebSocket\ConnectionException $e) {
$this->logger->error('Error while trying to send options.', ['code' => $e->getCode(), 'message' => $e->getMessage()]); $this->logger->error('Error while trying to send options.', ['code' => $e->getCode(), 'message' => $e->getMessage(), 'file' => $e->getFile(), 'line' => $e->getLine()]);
} catch (\ErrorException $e) {
$this->logger->notice('Warning while trying to send options.', ['code' => $e->getCode(), 'severity' => $e->getSeverity(), 'message' => $e->getMessage(), 'file' => $e->getFile(), 'line' => $e->getLine()]);
} }
} }