|
7 | 7 | import java.util.concurrent.ConcurrentHashMap;
|
8 | 8 | import java.util.function.Predicate;
|
9 | 9 | import java.util.stream.Stream;
|
10 |
| -import java.util.function.Predicate; |
11 |
| -import java.util.stream.Collectors; |
12 |
| -import java.util.stream.Stream; |
13 | 10 |
|
14 | 11 | import org.slf4j.Logger;
|
15 | 12 | import org.slf4j.LoggerFactory;
|
@@ -188,41 +185,31 @@ public Stream<T> list(Predicate<T> predicate) {
|
188 | 185 | return sharedIndexInformers.values().stream()
|
189 | 186 | .flatMap(i -> i.getStore().list().stream().filter(predicate));
|
190 | 187 | }
|
191 |
| - public Stream<T> getCachedCustomResources() { |
192 |
| - return getCachedCustomResources(a -> true); |
193 |
| - } |
194 | 188 |
|
195 |
| - public Stream<T> getCachedCustomResources(Predicate<T> predicate) { |
196 |
| - var streamList = sharedIndexInformers.values().stream() |
197 |
| - .map(i -> i.getStore().list().stream().filter(predicate)).collect(Collectors.toList()); |
198 |
| - if (streamList.size() == 1) { |
199 |
| - return streamList.get(0); |
200 |
| - } |
201 |
| - var resStream = streamList.get(0); |
202 |
| - for (int i = 1; i < streamList.size(); i++) { |
203 |
| - resStream = Stream.concat(resStream, streamList.get(i)); |
| 189 | + @Override |
| 190 | + public Stream<T> list(String namespace) { |
| 191 | + if (isWatchingAllNamespaces()) { |
| 192 | + return sharedIndexInformers.get(ANY_NAMESPACE_MAP_KEY).getStore().list().stream() |
| 193 | + .filter(r -> r.getMetadata() |
| 194 | + .getNamespace().equals(namespace)); |
| 195 | + } else { |
| 196 | + return sharedIndexInformers.get(namespace).getStore().list().stream(); |
204 | 197 | }
|
205 |
| - return resStream; |
206 | 198 | }
|
207 | 199 |
|
208 |
| - /** |
209 |
| - * Use only if the operator is watching specific namespace(s). |
210 |
| - * |
211 |
| - * @param namespace - namespace to filter on |
212 |
| - * @return stream of cached resources from the specified namespace |
213 |
| - */ |
214 |
| - public Stream<T> getCachedResources(String namespace) { |
215 |
| - return sharedIndexInformers.get(namespace).getStore().list().stream(); |
| 200 | + @Override |
| 201 | + public Stream<T> list(String namespace, Predicate<T> predicate) { |
| 202 | + if (isWatchingAllNamespaces()) { |
| 203 | + return sharedIndexInformers.get(ANY_NAMESPACE_MAP_KEY).getStore().list().stream() |
| 204 | + .filter(r -> r.getMetadata() |
| 205 | + .getNamespace().equals(namespace) && predicate.test(r)); |
| 206 | + } else { |
| 207 | + return sharedIndexInformers.get(namespace).getStore().list().stream().filter(predicate); |
| 208 | + } |
216 | 209 | }
|
217 | 210 |
|
218 |
| - /** |
219 |
| - * Use only if the operator is watching specific namespace(s). |
220 |
| - * |
221 |
| - * @param namespace - namespace to filter on |
222 |
| - * @return stream of cached resources from the specified namespace |
223 |
| - */ |
224 |
| - public Stream<T> getCachedResources(String namespace, Predicate<T> predicate) { |
225 |
| - return sharedIndexInformers.get(namespace).getStore().list().stream().filter(predicate); |
| 211 | + private boolean isWatchingAllNamespaces() { |
| 212 | + return sharedIndexInformers.containsKey(ANY_NAMESPACE_MAP_KEY); |
226 | 213 | }
|
227 | 214 |
|
228 | 215 | /**
|
|
0 commit comments