@@ -34,25 +34,21 @@ type sagaInvocation struct {
34
34
startedByRPCID string
35
35
}
36
36
37
- func (si * sagaInvocation ) setCorrelationIDs (message * gbus.BusMessage , isEvent bool , targetService string ) {
37
+ func (si * sagaInvocation ) setCorrelationIDs (message * gbus.BusMessage , targetService string , semantics gbus. Semantics ) {
38
38
39
- message .CorrelationID = si .inboundMsg .ID
40
39
message .SagaID = si .sagaID
41
40
42
- if ! isEvent {
43
- //support saga-to-saga communication
44
- if si .inboundMsg .SagaID != "" {
45
- message . SagaCorrelationID = si . inboundMsg . SagaID
46
- }
41
+ if semantics == gbus . REPLY {
42
+ message . CorrelationID = si . inboundMsg . ID
43
+ message . SagaCorrelationID = si .inboundMsg .SagaID
44
+
45
+ } else if semantics == gbus . CMD {
47
46
//if the saga is potentially invoking itself then set the SagaCorrelationID to reflect that
48
47
//https://github.com/wework/grabbit/issues/64
49
-
50
48
if targetService == si .hostingSvc {
51
49
message .SagaCorrelationID = message .SagaID
52
50
}
53
-
54
51
}
55
-
56
52
}
57
53
func (si * sagaInvocation ) HostingSvc () string {
58
54
return si .hostingSvc
@@ -64,13 +60,13 @@ func (si *sagaInvocation) InvokingSvc() string {
64
60
65
61
func (si * sagaInvocation ) Reply (ctx context.Context , message * gbus.BusMessage ) error {
66
62
_ , targetService := si .decoratedInvocation .Routing ()
67
- si .setCorrelationIDs (message , false , targetService )
63
+ si .setCorrelationIDs (message , targetService , gbus . REPLY )
68
64
return si .decoratedInvocation .Reply (ctx , message )
69
65
}
70
66
71
67
func (si * sagaInvocation ) ReplyToInitiator (ctx context.Context , message * gbus.BusMessage ) error {
72
68
73
- si .setCorrelationIDs (message , false , si .startedBy )
69
+ si .setCorrelationIDs (message , si .startedBy , gbus . REPLY )
74
70
75
71
//override the correlation ids to those of the message creating the saga
76
72
message .SagaCorrelationID = si .startedBySaga
@@ -93,13 +89,13 @@ func (si *sagaInvocation) Ctx() context.Context {
93
89
94
90
func (si * sagaInvocation ) Send (ctx context.Context , toService string ,
95
91
command * gbus.BusMessage , policies ... gbus.MessagePolicy ) error {
96
- si .setCorrelationIDs (command , false , toService )
92
+ si .setCorrelationIDs (command , toService , gbus . CMD )
97
93
return si .decoratedBus .Send (ctx , toService , command , policies ... )
98
94
}
99
95
100
96
func (si * sagaInvocation ) Publish (ctx context.Context , exchange , topic string ,
101
97
event * gbus.BusMessage , policies ... gbus.MessagePolicy ) error {
102
- si .setCorrelationIDs (event , true , "" )
98
+ si .setCorrelationIDs (event , "" , gbus . EVT )
103
99
return si .decoratedBus .Publish (ctx , exchange , topic , event , policies ... )
104
100
}
105
101
0 commit comments