1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
|
<?php
declare(strict_types=1);
namespace OpenTelemetry\SDK\Common\Export\Stream;
use BadMethodCallException;
use ErrorException;
use function fflush;
use function fwrite;
use OpenTelemetry\SDK\Common\Export\TransportInterface;
use OpenTelemetry\SDK\Common\Future\CancellationInterface;
use OpenTelemetry\SDK\Common\Future\CompletedFuture;
use OpenTelemetry\SDK\Common\Future\ErrorFuture;
use OpenTelemetry\SDK\Common\Future\FutureInterface;
use function restore_error_handler;
use RuntimeException;
use function set_error_handler;
use function strlen;
use Throwable;
/**
* @internal
*
* @psalm-template CONTENT_TYPE of string
* @template-implements TransportInterface<CONTENT_TYPE>
*/
final class StreamTransport implements TransportInterface
{
/**
* @var resource|null
*/
private $stream;
private string $contentType;
/**
* @param resource $stream
*
* @psalm-param CONTENT_TYPE $contentType
*/
public function __construct($stream, string $contentType)
{
$this->stream = $stream;
$this->contentType = $contentType;
}
public function contentType(): string
{
return $this->contentType;
}
public function send(string $payload, ?CancellationInterface $cancellation = null): FutureInterface
{
if (!$this->stream) {
return new ErrorFuture(new BadMethodCallException('Transport closed'));
}
set_error_handler(static function (int $errno, string $errstr, string $errfile, int $errline): bool {
throw new ErrorException($errstr, 0, $errno, $errfile, $errline);
});
try {
$bytesWritten = fwrite($this->stream, $payload);
} catch (Throwable $e) {
return new ErrorFuture($e);
} finally {
restore_error_handler();
}
if ($bytesWritten !== strlen($payload)) {
return new ErrorFuture(new RuntimeException(sprintf('Write failure, wrote %d of %d bytes', $bytesWritten, strlen($payload))));
}
return new CompletedFuture(null);
}
public function shutdown(?CancellationInterface $cancellation = null): bool
{
if (!$this->stream) {
return false;
}
$flush = @fflush($this->stream);
$this->stream = null;
return $flush;
}
public function forceFlush(?CancellationInterface $cancellation = null): bool
{
if (!$this->stream) {
return false;
}
return @fflush($this->stream);
}
}
|
