|
1
|
|
|
<?php |
|
2
|
|
|
namespace PSB\Core\Pipeline; |
|
3
|
|
|
|
|
4
|
|
|
|
|
5
|
|
|
use PSB\Core\OutgoingOptions; |
|
6
|
|
|
use PSB\Core\Pipeline\Incoming\IncomingContext; |
|
7
|
|
|
use PSB\Core\PublishOptions; |
|
8
|
|
|
use PSB\Core\ReplyOptions; |
|
9
|
|
|
use PSB\Core\SendOptions; |
|
10
|
|
|
use PSB\Core\SubscribeOptions; |
|
11
|
|
|
use PSB\Core\UnsubscribeOptions; |
|
12
|
|
|
use PSB\Core\UuidGeneration\UuidGeneratorInterface; |
|
13
|
|
|
|
|
14
|
|
|
class BusOperations |
|
15
|
|
|
{ |
|
16
|
|
|
/** |
|
17
|
|
|
* @var PipelineFactory |
|
18
|
|
|
*/ |
|
19
|
|
|
private $pipelineFactory; |
|
20
|
|
|
|
|
21
|
|
|
/** |
|
22
|
|
|
* @var BusOperationsContextFactory |
|
23
|
|
|
*/ |
|
24
|
|
|
private $busOperationsContextFactory; |
|
25
|
|
|
|
|
26
|
|
|
/** |
|
27
|
|
|
* @var PipelineModifications |
|
28
|
|
|
*/ |
|
29
|
|
|
private $pipelineModifications; |
|
30
|
|
|
|
|
31
|
|
|
/** |
|
32
|
|
|
* @var UuidGeneratorInterface |
|
33
|
|
|
*/ |
|
34
|
|
|
private $uuidGenerator; |
|
35
|
|
|
|
|
36
|
|
|
/** |
|
37
|
|
|
* @param PipelineFactory $pipelineFactory |
|
38
|
|
|
* @param BusOperationsContextFactory $busOperationsContextFactory |
|
39
|
|
|
* @param PipelineModifications $pipelineModifications |
|
40
|
|
|
* @param UuidGeneratorInterface $uuidGenerator |
|
41
|
|
|
*/ |
|
42
|
10 |
|
public function __construct( |
|
43
|
|
|
PipelineFactory $pipelineFactory, |
|
44
|
|
|
BusOperationsContextFactory $busOperationsContextFactory, |
|
45
|
|
|
PipelineModifications $pipelineModifications, |
|
46
|
|
|
UuidGeneratorInterface $uuidGenerator |
|
47
|
|
|
) { |
|
48
|
10 |
|
$this->pipelineFactory = $pipelineFactory; |
|
49
|
10 |
|
$this->busOperationsContextFactory = $busOperationsContextFactory; |
|
50
|
10 |
|
$this->pipelineModifications = $pipelineModifications; |
|
51
|
10 |
|
$this->uuidGenerator = $uuidGenerator; |
|
52
|
10 |
|
} |
|
53
|
|
|
|
|
54
|
|
|
/** |
|
55
|
|
|
* @param object $message |
|
56
|
|
|
* @param SendOptions $options |
|
57
|
|
|
* @param PipelineStageContext $parentContext |
|
58
|
|
|
*/ |
|
59
|
3 |
View Code Duplication |
public function send($message, SendOptions $options, PipelineStageContext $parentContext) |
|
|
|
|
|
|
60
|
|
|
{ |
|
61
|
3 |
|
$this->ensureMessageId($options); |
|
62
|
3 |
|
$sendContext = $this->busOperationsContextFactory->createSendContext($message, $options, $parentContext); |
|
63
|
3 |
|
$pipeline = $this->pipelineFactory->createStartingWith(get_class($sendContext), $this->pipelineModifications); |
|
64
|
|
|
|
|
65
|
3 |
|
$pipeline->invoke($sendContext); |
|
66
|
3 |
|
} |
|
67
|
|
|
|
|
68
|
|
|
/** |
|
69
|
|
|
* @param object $message |
|
70
|
|
|
* @param SendOptions $options |
|
71
|
|
|
* @param PipelineStageContext $parentContext |
|
72
|
|
|
*/ |
|
73
|
1 |
|
public function sendLocal($message, SendOptions $options, PipelineStageContext $parentContext) |
|
74
|
|
|
{ |
|
75
|
1 |
|
$options->routeToLocalEndpointInstance(); |
|
76
|
|
|
|
|
77
|
1 |
|
$this->send($message, $options, $parentContext); |
|
78
|
1 |
|
} |
|
79
|
|
|
|
|
80
|
|
|
/** |
|
81
|
|
|
* @param object $message |
|
82
|
|
|
* @param PublishOptions $options |
|
83
|
|
|
* @param PipelineStageContext $parentContext |
|
84
|
|
|
*/ |
|
85
|
2 |
View Code Duplication |
public function publish($message, PublishOptions $options, PipelineStageContext $parentContext) |
|
|
|
|
|
|
86
|
|
|
{ |
|
87
|
2 |
|
$this->ensureMessageId($options); |
|
88
|
2 |
|
$publishContext = $this->busOperationsContextFactory->createPublishContext($message, $options, $parentContext); |
|
89
|
2 |
|
$pipeline = $this->pipelineFactory->createStartingWith( |
|
90
|
2 |
|
get_class($publishContext), |
|
91
|
2 |
|
$this->pipelineModifications |
|
92
|
2 |
|
); |
|
93
|
|
|
|
|
94
|
2 |
|
$pipeline->invoke($publishContext); |
|
95
|
2 |
|
} |
|
96
|
|
|
|
|
97
|
|
|
/** |
|
98
|
|
|
* @param object $message |
|
99
|
|
|
* @param ReplyOptions $options |
|
100
|
|
|
* @param IncomingContext $parentContext |
|
101
|
|
|
*/ |
|
102
|
2 |
View Code Duplication |
public function reply($message, ReplyOptions $options, IncomingContext $parentContext) |
|
|
|
|
|
|
103
|
|
|
{ |
|
104
|
2 |
|
$this->ensureMessageId($options); |
|
105
|
2 |
|
$publishContext = $this->busOperationsContextFactory->createReplyContext($message, $options, $parentContext); |
|
106
|
2 |
|
$pipeline = $this->pipelineFactory->createStartingWith( |
|
107
|
2 |
|
get_class($publishContext), |
|
108
|
2 |
|
$this->pipelineModifications |
|
109
|
2 |
|
); |
|
110
|
|
|
|
|
111
|
2 |
|
$pipeline->invoke($publishContext); |
|
112
|
2 |
|
} |
|
113
|
|
|
|
|
114
|
|
|
/** |
|
115
|
|
|
* @param string $eventFqcn |
|
116
|
|
|
* @param SubscribeOptions $options |
|
117
|
|
|
* @param PipelineStageContext $parentContext |
|
118
|
|
|
*/ |
|
119
|
1 |
View Code Duplication |
public function subscribe($eventFqcn, SubscribeOptions $options, PipelineStageContext $parentContext) |
|
|
|
|
|
|
120
|
|
|
{ |
|
121
|
1 |
|
$subscribeContext = $this->busOperationsContextFactory->createSubscribeContext( |
|
122
|
1 |
|
$eventFqcn, |
|
123
|
1 |
|
$options, |
|
124
|
|
|
$parentContext |
|
125
|
1 |
|
); |
|
126
|
1 |
|
$pipeline = $this->pipelineFactory->createStartingWith( |
|
127
|
1 |
|
get_class($subscribeContext), |
|
128
|
1 |
|
$this->pipelineModifications |
|
129
|
1 |
|
); |
|
130
|
|
|
|
|
131
|
1 |
|
$pipeline->invoke($subscribeContext); |
|
132
|
1 |
|
} |
|
133
|
|
|
|
|
134
|
|
|
/** |
|
135
|
|
|
* @param string $eventFqcn |
|
136
|
|
|
* @param UnsubscribeOptions $options |
|
137
|
|
|
* @param PipelineStageContext $parentContext |
|
138
|
|
|
*/ |
|
139
|
1 |
View Code Duplication |
public function unsubscribe($eventFqcn, UnsubscribeOptions $options, PipelineStageContext $parentContext) |
|
|
|
|
|
|
140
|
|
|
{ |
|
141
|
1 |
|
$unsubscribeContext = $this->busOperationsContextFactory->createUnsubscribeContext( |
|
142
|
1 |
|
$eventFqcn, |
|
143
|
1 |
|
$options, |
|
144
|
|
|
$parentContext |
|
145
|
1 |
|
); |
|
146
|
1 |
|
$pipeline = $this->pipelineFactory->createStartingWith( |
|
147
|
1 |
|
get_class($unsubscribeContext), |
|
148
|
1 |
|
$this->pipelineModifications |
|
149
|
1 |
|
); |
|
150
|
|
|
|
|
151
|
1 |
|
$pipeline->invoke($unsubscribeContext); |
|
152
|
1 |
|
} |
|
153
|
|
|
|
|
154
|
|
|
/** |
|
155
|
|
|
* @param OutgoingOptions $options |
|
156
|
|
|
*/ |
|
157
|
7 |
|
private function ensureMessageId(OutgoingOptions $options) |
|
158
|
|
|
{ |
|
159
|
7 |
|
if (!$options->getMessageId()) { |
|
160
|
3 |
|
$options->setMessageId($this->uuidGenerator->generate()); |
|
161
|
3 |
|
} |
|
162
|
7 |
|
} |
|
163
|
|
|
} |
|
164
|
|
|
|
ghola /
core
Loading history...
Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation.
You can also find more detailed suggestions in the “Code” section of your repository.