enqueue-dev/pkg/enqueue/Client/TraceableProducer.php at master · php-enqueue/enqueue-dev · GitHub

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
98
99
100
101
102
103
104
105
<?php

namespace Enqueue\Client;

use Enqueue\Rpc\Promise;

final class TraceableProducer implements ProducerInterface
{
    /**
     * @var array
     */
    private $traces = [];

    /**
     * @var ProducerInterface
     */
    private $producer;

    public function __construct(ProducerInterface $producer)
    {
        $this->producer = $producer;
    }

    public function sendEvent(string $topic, $message): void
    {
        $this->producer->sendEvent($topic, $message);

        $this->collectTrace($topic, null, $message);
    }

    public function sendCommand(string $command, $message, bool $needReply = false): ?Promise
    {
        $result = $this->producer->sendCommand($command, $message, $needReply);

        $this->collectTrace(null, $command, $message);

        return $result;
    }

    public function getTopicTraces(string $topic): array
    {
        $topicTraces = [];
        foreach ($this->traces as $trace) {
            if ($topic == $trace['topic']) {
                $topicTraces[] = $trace;
            }
        }

        return $topicTraces;
    }

    public function getCommandTraces(string $command): array
    {
        $commandTraces = [];
        foreach ($this->traces as $trace) {
            if ($command == $trace['command']) {
                $commandTraces[] = $trace;
            }
        }

        return $commandTraces;
    }

    public function getTraces(): array
    {
        return $this->traces;
    }

    public function clearTraces(): void
    {
        $this->traces = [];
    }

    private function collectTrace(?string $topic, ?string $command, $message): void
    {
        $trace = [
            'topic' => $topic,
            'command' => $command,
            'body' => $message,
            'headers' => [],
            'properties' => [],
            'priority' => null,
            'expire' => null,
            'delay' => null,
            'timestamp' => null,
            'contentType' => null,
            'messageId' => null,
            'sentAt' => (new \DateTime())->format('Y-m-d H:i:s.u'),
        ];

        if ($message instanceof Message) {
            $trace['body'] = $message->getBody();
            $trace['headers'] = $message->getHeaders();
            $trace['properties'] = $message->getProperties();
            $trace['priority'] = $message->getPriority();
            $trace['expire'] = $message->getExpire();
            $trace['delay'] = $message->getDelay();
            $trace['timestamp'] = $message->getTimestamp();
            $trace['contentType'] = $message->getContentType();
            $trace['messageId'] = $message->getMessageId();
        }

        $this->traces[] = $trace;
    }
}