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
|
/*-
* ============LICENSE_START=======================================================
* ONAP Policy Models
* ================================================================================
* Copyright (C) 2019, 2021 AT&T Intellectual Property. All rights reserved.
* ================================================================================
* 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.
* ============LICENSE_END=========================================================
*/
package org.onap.policy.models.sim.dmaap.provider;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Data associated with a consumer group. All consumer instances within a group share the
* same data object.
*/
public class ConsumerGroupData {
private static final Logger logger = LoggerFactory.getLogger(ConsumerGroupData.class);
/**
* Returned when messages can no longer be read from this consumer group object,
* because it is being removed from the topic. {@link #UNREADABLE_LIST} must not be
* the same list as Collections.emptyList(), thus we wrap it.
*/
public static final List<String> UNREADABLE_LIST = Collections.unmodifiableList(Collections.emptyList());
/**
* Returned when there are no messages read. Collections.emptyList() is already
* unmodifiable, thus no need to wrap it.
*/
private static final List<String> EMPTY_LIST = Collections.emptyList();
/**
* This is locked while fields other than {@link #messageQueue} are updated.
*/
private final Object lockit = new Object();
/**
* Number of sweep cycles that have occurred since a consumer has attempted to read
* from the queue. This consumer group should be removed once this count exceeds
* {@code 1}, provided {@link #nreaders} is zero.
*/
private int nsweeps = 0;
/**
* Number of consumers that are currently attempting to read from the queue. This
* consumer group should not be removed as long as this is non-zero.
*/
private int nreaders = 0;
/**
* Message queue.
*/
private final BlockingQueue<String> messageQueue = new LinkedBlockingQueue<>();
/**
* Constructs the object.
*
* @param topicName name of the topic with which this object is associated
* @param groupName name of the consumer group with which this object is associated
*/
public ConsumerGroupData(String topicName, String groupName) {
logger.info("Topic {}: add consumer group: {}", topicName, groupName);
}
/**
* Determines if this consumer group should be removed. This should be invoked once
* during each sweep cycle. When this returns {@code true}, this consumer group should
* be immediately discarded, as any readers will sit in a spin loop waiting for it to
* be discarded.
*
* @return {@code true} if this consumer group should be removed, {@code false}
* otherwise
*/
public boolean shouldRemove() {
synchronized (lockit) {
return (nreaders == 0 && ++nsweeps > 1);
}
}
/**
* Reads messages from the queue, blocking if necessary.
*
* @param maxRead maximum number of messages to read
* @param waitMs time to wait, in milliseconds, if the queue is currently empty
* @return a list of messages read from the queue, empty if no messages became
* available before the wait time elapsed, or {@link #UNREADABLE_LIST} if this
* consumer group object is no longer active
* @throws InterruptedException if this thread was interrupted while waiting for the
* first message
*/
public List<String> read(int maxRead, long waitMs) throws InterruptedException {
synchronized (lockit) {
if (nsweeps > 1 && nreaders == 0) {
// cannot use this consumer group object anymore
return UNREADABLE_LIST;
}
++nreaders;
}
/*
* Note: do EVERYTHING inside of the "try" block, so that the "finally" block can
* update the reader count.
*
* Do NOT hold the lockit while we're polling, as poll() may block for a while.
*/
try {
// always read at least one message
int maxRead2 = Math.max(1, maxRead);
long waitMs2 = Math.max(0, waitMs);
// perform a blocking read of the queue
String obj = getNextMessage(waitMs2);
if (obj == null) {
return EMPTY_LIST;
}
/*
* List should hold all messages from the queue PLUS the one we already have.
* Note: it's possible for additional messages to be added to the queue while
* we're reading from it. In that case, the list will grow as needed.
*/
List<String> lst = new ArrayList<>(Math.min(maxRead2, messageQueue.size() + 1));
lst.add(obj);
// perform NON-blocking read of subsequent messages
for (var x = 1; x < maxRead2; ++x) {
if ((obj = messageQueue.poll()) == null) {
break;
}
lst.add(obj);
}
return lst;
} finally {
synchronized (lockit) {
--nreaders;
nsweeps = 0;
}
}
}
/**
* Writes messages to the queue.
*
* @param messages messages to be written to the queue
*/
public void write(List<String> messages) {
messageQueue.addAll(messages);
}
// the following methods may be overridden by junit tests
/**
* Gets the next message from the queue.
*
* @param waitMs time to wait, in milliseconds, if the queue is currently empty
* @return the next message, or {@code null} if no messages became available before
* the wait time elapsed
* @throws InterruptedException if this thread was interrupted while waiting for the
* message
*/
protected String getNextMessage(long waitMs) throws InterruptedException {
return messageQueue.poll(waitMs, TimeUnit.MILLISECONDS);
}
}
|
