forked from lirm/aeron-go
/
subscription.go
405 lines (348 loc) · 12.7 KB
/
subscription.go
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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
/*
Copyright 2016 Stanislav Liberman
Copyright (C) 2022 Talos, Inc.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package aeron
import (
"errors"
ctr "github.com/lirm/aeron-go/aeron/counters"
"strings"
"github.com/lirm/aeron-go/aeron/atomic"
"github.com/lirm/aeron-go/aeron/logbuffer/term"
)
const (
ChannelStatusNoIdAllocated = -1 // Channel status counter not allocated for IPC channels
ChannelStatusErrored = -1 // Channel has errored. Check logs for information
ChannelStatusInitializing = 0 // Channel is being initialized
ChannelStatusActive = 1 // Channel has finished initialization and is active
ChannelStatusClosing = 2 // Channel is being closed
)
// ChannelStatusString provides a convenience method for logging and error handling
func ChannelStatusString(channelStatus int) string {
switch channelStatus {
case ChannelStatusErrored:
return "ChannelStatusErrored"
case ChannelStatusInitializing:
return "ChannelStatusInitializing"
case ChannelStatusActive:
return "ChannelStatusActive"
case ChannelStatusClosing:
return "ChannelStatusClosing"
default:
return "Unknown"
}
}
// From LocalSocketAddressStatus.Java
const (
ChannelStatusIdOffset = 0
LocalSocketAddressLengthOffset = ChannelStatusIdOffset + 4
LocalSocketAddressStringOffset = LocalSocketAddressLengthOffset + 4
)
const LocalSocketAddressStatusCounterTypeId = 14
type ReceivingConductor interface {
CounterReader() *ctr.Reader
releaseSubscription(regID int64, images []Image) error
AddRcvDestination(registrationID int64, endpointChannel string) error
RemoveRcvDestination(registrationID int64, endpointChannel string) error
}
// Subscription is the object responsible for receiving messages from media driver. It is specific to a channel and
// stream ID combination.
type Subscription struct {
conductor ReceivingConductor
channel string
roundRobinIndex int
registrationID int64
streamID int32
channelStatusID int32
images *ImageList
isClosed atomic.Bool
availableImageHandler AvailableImageHandler
unavailableImageHandler UnavailableImageHandler
}
// NewSubscription is a factory method to create new subscription to be added to the media driver
func NewSubscription(
conductor ReceivingConductor,
channel string,
registrationID int64,
streamID int32,
channelStatusID int32,
availableImagehandler AvailableImageHandler,
unavailableImageHandler UnavailableImageHandler) *Subscription {
sub := new(Subscription)
sub.images = NewImageList()
sub.conductor = conductor
sub.channel = channel
sub.registrationID = registrationID
sub.streamID = streamID
sub.channelStatusID = channelStatusID
sub.roundRobinIndex = 0
sub.isClosed.Set(false)
sub.availableImageHandler = availableImagehandler
sub.unavailableImageHandler = unavailableImageHandler
return sub
}
// Channel returns the media address for delivery to the channel.
func (sub *Subscription) Channel() string {
return sub.channel
}
// StreamID returns Stream identity for scoping within the channel media address.
func (sub *Subscription) StreamID() int32 {
return sub.streamID
}
// IsClosed returns whether this subscription has been closed.
func (sub *Subscription) IsClosed() bool {
return sub.isClosed.Get()
}
// ChannelStatus returns the status of the media channel for this Subscription.
// The status will be ChannelStatusErrored if a socket exception on setup or ChannelStatusActive if all is well.
func (sub *Subscription) ChannelStatus() int {
if sub.IsClosed() {
return ChannelStatusNoIdAllocated
}
if sub.channelStatusID == -1 { // IPC channels don't have a channel status counter
return ChannelStatusActive
}
return int(sub.conductor.CounterReader().GetCounterValue(sub.channelStatusID))
}
// ChannelStatusId returns the counter ID used to represent the channel status of this Subscription.
func (sub *Subscription) ChannelStatusId() int32 {
return sub.channelStatusID
}
// Close will release all images in this subscription, send command to the driver and block waiting for response from
// the media driver. Images will be lingered by the ClientConductor.
func (sub *Subscription) Close() error {
if sub.isClosed.CompareAndSet(false, true) {
images := sub.images.Empty()
return sub.conductor.releaseSubscription(sub.registrationID, images)
}
return nil
}
// AvailableImageHandler returns a callback used to indicate when an Image becomes available under this Subscription.
// The handler may be nil.
func (sub *Subscription) AvailableImageHandler() AvailableImageHandler {
return sub.availableImageHandler
}
// UnavailableImageHandler returns a callback used to indicate when an Image goes unavailable under this Subscription.
// The handler may be nil.
func (sub *Subscription) UnavailableImageHandler() UnavailableImageHandler {
return sub.unavailableImageHandler
}
// Poll is the primary receive mechanism on subscription.
func (sub *Subscription) Poll(handler term.FragmentHandler, fragmentLimit int) int {
img := sub.images.Get()
length := len(img)
var fragmentsRead int
if length > 0 {
startingIndex := sub.roundRobinIndex
sub.roundRobinIndex++
if startingIndex >= length {
sub.roundRobinIndex = 0
startingIndex = 0
}
for i := startingIndex; i < length && fragmentsRead < fragmentLimit; i++ {
var newFrags int
newFrags = img[i].Poll(handler, fragmentLimit-fragmentsRead)
fragmentsRead += newFrags
}
for i := 0; i < startingIndex && fragmentsRead < fragmentLimit; i++ {
var newFrags int
newFrags = img[i].Poll(handler, fragmentLimit-fragmentsRead)
fragmentsRead += newFrags
}
}
return fragmentsRead
}
// ControlledPoll polls in a controlled manner the image s under the subscription for available message fragments.
// Control is applied to fragments in the stream. If more fragments can be read on another stream
// they will even if BREAK or ABORT is returned from the fragment handler.
//
// Each fragment read will be a whole message if it is under MTU length. If larger than MTU then it will come
// as a series of fragments ordered within a session.
// Returns the number of fragments received.
func (sub *Subscription) ControlledPoll(handler term.ControlledFragmentHandler, fragmentLimit int) int {
img := sub.images.Get()
length := len(img)
var fragmentsRead int
if length > 0 {
startingIndex := sub.roundRobinIndex
sub.roundRobinIndex++
if startingIndex >= length {
sub.roundRobinIndex = 0
startingIndex = 0
}
for i := startingIndex; i < length && fragmentsRead < fragmentLimit; i++ {
var newFrags int
newFrags = img[i].ControlledPoll(handler, fragmentLimit-fragmentsRead)
fragmentsRead += newFrags
}
for i := 0; i < startingIndex && fragmentsRead < fragmentLimit; i++ {
var newFrags int
newFrags = img[i].ControlledPoll(handler, fragmentLimit-fragmentsRead)
fragmentsRead += newFrags
}
}
return fragmentsRead
}
//go:norace
func (sub *Subscription) hasImage(sessionID int32) bool {
img := sub.images.Get()
for _, image := range img {
if image.SessionID() == sessionID {
return true
}
}
return false
}
//go:norace
func (sub *Subscription) addImage(image Image) *[]Image {
images := sub.images.Get()
sub.images.Set(append(images, image))
return &images
}
//go:norace
func (sub *Subscription) removeImage(correlationID int64) Image {
img := sub.images.Get()
for ix, image := range img {
if image.CorrelationID() == correlationID {
logger.Debugf("Removing image %v for subscription %d", image, sub.registrationID)
img[ix] = img[len(img)-1]
img = img[:len(img)-1]
sub.images.Set(img)
return image
}
}
return nil
}
// RegistrationID returns the registration id.
func (sub *Subscription) RegistrationID() int64 {
return sub.registrationID
}
// IsConnected returns if this subscription is connected by having at least one open publication image.
func (sub *Subscription) IsConnected() bool {
for _, image := range sub.images.Get() {
if !image.IsClosed() {
return true
}
}
return false
}
// HasImages is a helper method checking whether this subscription has any images associated with it.
func (sub *Subscription) HasImages() bool {
images := sub.images.Get()
return len(images) > 0
}
// ImageCount count of images associated with this subscription.
func (sub *Subscription) ImageCount() int {
images := sub.images.Get()
return len(images)
}
// ImageBySessionID returns the associated with the given sessionId.
func (sub *Subscription) ImageBySessionID(sessionID int32) Image {
img := sub.images.Get()
for _, image := range img {
if image.SessionID() == sessionID {
return image
}
}
return nil
}
// ResolvedEndpoint finds the resolved endpoint for the channel. This
// may be nil if MDS is used and no destination is yet added.
// The result is simply the first in the list of addresses found if
// multiple addresses exist
func (sub *Subscription) ResolvedEndpoint() string {
reader := sub.conductor.CounterReader()
if sub.ChannelStatus() != ChannelStatusActive {
return ""
}
var endpoint string
reader.ScanForType(LocalSocketAddressStatusCounterTypeId, func(counterId int32, keyBuffer *atomic.Buffer) bool {
channelStatusId := keyBuffer.GetInt32(ChannelStatusIdOffset)
length := keyBuffer.GetInt32(LocalSocketAddressLengthOffset)
if channelStatusId == sub.channelStatusID && length > 0 && reader.GetCounterValue(counterId) == ChannelStatusActive {
endpoint = string(keyBuffer.GetBytesArray(LocalSocketAddressStringOffset, length))
return false
}
return true
})
return endpoint
}
// TryResolveChannelEndpointPort resolves the channel endpoint and replaces it with the port from the
// ephemeral range when 0 was provided. If there are no addresses, or if there is more than one, returned from
// LocalSocketAddresses() then the original channel is returned.
// If the channel is not ACTIVE, then empty string will be returned.
func (sub *Subscription) TryResolveChannelEndpointPort() string {
if sub.ChannelStatus() != ChannelStatusActive {
return ""
}
localSocketAddresses := sub.LocalSocketAddresses()
if len(localSocketAddresses) != 1 {
return sub.channel
}
uri, err := ParseChannelUri(sub.channel)
if err != nil {
logger.Warningf("error parsing channel (%s): %v", sub.channel, err)
return sub.channel
}
endpoint := uri.Get("endpoint")
if strings.HasSuffix(endpoint, ":0") {
resolvedEndpoint := localSocketAddresses[0]
i := strings.LastIndex(resolvedEndpoint, ":")
uri.Set("endpoint", endpoint[:(len(endpoint)-2)]+resolvedEndpoint[i:])
return uri.String()
}
return sub.channel
}
// LocalSocketAddresses fetches the local socket addresses for this subscription.
func (sub *Subscription) LocalSocketAddresses() []string {
if sub.ChannelStatus() != ChannelStatusActive {
return nil
}
var bindings []string
reader := sub.conductor.CounterReader()
reader.ScanForType(LocalSocketAddressStatusCounterTypeId, func(counterId int32, keyBuffer *atomic.Buffer) bool {
channelStatusId := keyBuffer.GetInt32(ChannelStatusIdOffset)
length := keyBuffer.GetInt32(LocalSocketAddressLengthOffset)
if channelStatusId == sub.channelStatusID && length > 0 && reader.GetCounterValue(counterId) == ChannelStatusActive {
bindings = append(bindings, string(keyBuffer.GetBytesArray(LocalSocketAddressStringOffset, length)))
}
return true
})
return bindings
}
// AddDestination adds a destination manually to a multi-destination Subscription.
func (sub *Subscription) AddDestination(endpointChannel string) error {
if sub.IsClosed() {
return errors.New("subscription is closed")
}
return sub.conductor.AddRcvDestination(sub.registrationID, endpointChannel)
}
// RemoveDestination removes a destination manually from a multi-destination Subscription.
func (sub *Subscription) RemoveDestination(endpointChannel string) error {
if sub.IsClosed() {
return errors.New("subscription is closed")
}
return sub.conductor.RemoveRcvDestination(sub.registrationID, endpointChannel)
}
// IsConnectedTo is a helper function used primarily by tests, which is used within the same process to verify that
// subscription is connected to a specific publication.
func IsConnectedTo(sub *Subscription, pub *Publication) bool {
img := sub.images.Get()
if sub.channel == pub.channel && sub.streamID == pub.streamID {
for _, image := range img {
if image.SessionID() == pub.sessionID {
return true
}
}
}
return false
}