@Test void testAddStreamSubscription() { final String streamName = StreamType.DEFAULT_EVENT_STREAM; Optional<Subscription> subscriptionO = SubscriptionFactory.getSubscription("subId1", "false"); assertTrue(subscriptionO.isPresent()); streamManager.addStreamSubscription(subscriptionO.get()); assertTrue(streamManager.hasSubscriptions(streamName)); Set<Subscription> streamSubscriptions = streamManager.getStreamSubscriptions(streamName); assertEquals(1, streamSubscriptions.size()); for (final Subscription s : streamSubscriptions) { assertEquals(s, subscriptionO.get()); } assertFalse(streamManager.getQueueForStream(streamName).isPresent()); assertFalse(streamManager.getStreamMetrics(streamName).isPresent()); assertTrue(streamManager.getRegisteredStreams().isEmpty()); }
synchronized void addStreamSubscription(final Subscription sub) { List<String> streams = sanitizeStreamSubjects(sub.getSubjects()); LOG.info("adding subscription {} with streams {}", sub, streams); handleDuplicateSubscriptionId(sub); for (String stream : streams) { streamSubscriptionsMap.putIfAbsent(stream, new ConcurrentSkipListSet<>()); ConcurrentSkipListSet<Subscription> subs = streamSubscriptionsMap.get(stream); int maxSubs = config.maxSubscriptions(stream); subs.removeIf(s -> s.getSubscriptionId().equals(sub.getSubscriptionId())); subs.add(sub); int numSubs = subs.size(); if (numSubs > maxSubs) { removeStreamSubscription(sub); LOG.warn("QUERY SUBSCRIPTION REJECTED: Number of subscriptions for stream {} exceeded max {}. " + "Increase (default mre.publish.max.subscriptions.per.stream.default or " + " mre.publish.max.subscriptions.stream.<streamName>) to allow more queries. Removed {}", stream, maxSubs, sub); getStreamMetrics(stream).ifPresent(m -> m.getMantisQueryRejectedCounter().increment()); } getStreamMetrics(stream).ifPresent(m -> m.getMantisActiveQueryCountGauge().set((double) numSubs)); } subscriptionIdToStreamsMap.put(sub.getSubscriptionId(), streams); }
StreamManager { synchronized void addStreamSubscription(final Subscription sub) { List<String> streams = sanitizeStreamSubjects(sub.getSubjects()); LOG.info("adding subscription {} with streams {}", sub, streams); handleDuplicateSubscriptionId(sub); for (String stream : streams) { streamSubscriptionsMap.putIfAbsent(stream, new ConcurrentSkipListSet<>()); ConcurrentSkipListSet<Subscription> subs = streamSubscriptionsMap.get(stream); int maxSubs = config.maxSubscriptions(stream); subs.removeIf(s -> s.getSubscriptionId().equals(sub.getSubscriptionId())); subs.add(sub); int numSubs = subs.size(); if (numSubs > maxSubs) { removeStreamSubscription(sub); LOG.warn("QUERY SUBSCRIPTION REJECTED: Number of subscriptions for stream {} exceeded max {}. " + "Increase (default mre.publish.max.subscriptions.per.stream.default or " + " mre.publish.max.subscriptions.stream.<streamName>) to allow more queries. Removed {}", stream, maxSubs, sub); getStreamMetrics(stream).ifPresent(m -> m.getMantisQueryRejectedCounter().increment()); } getStreamMetrics(stream).ifPresent(m -> m.getMantisActiveQueryCountGauge().set((double) numSubs)); } subscriptionIdToStreamsMap.put(sub.getSubscriptionId(), streams); } }
StreamManager { synchronized void addStreamSubscription(final Subscription sub) { List<String> streams = sanitizeStreamSubjects(sub.getSubjects()); LOG.info("adding subscription {} with streams {}", sub, streams); handleDuplicateSubscriptionId(sub); for (String stream : streams) { streamSubscriptionsMap.putIfAbsent(stream, new ConcurrentSkipListSet<>()); ConcurrentSkipListSet<Subscription> subs = streamSubscriptionsMap.get(stream); int maxSubs = config.maxSubscriptions(stream); subs.removeIf(s -> s.getSubscriptionId().equals(sub.getSubscriptionId())); subs.add(sub); int numSubs = subs.size(); if (numSubs > maxSubs) { removeStreamSubscription(sub); LOG.warn("QUERY SUBSCRIPTION REJECTED: Number of subscriptions for stream {} exceeded max {}. " + "Increase (default mre.publish.max.subscriptions.per.stream.default or " + " mre.publish.max.subscriptions.stream.<streamName>) to allow more queries. Removed {}", stream, maxSubs, sub); getStreamMetrics(stream).ifPresent(m -> m.getMantisQueryRejectedCounter().increment()); } getStreamMetrics(stream).ifPresent(m -> m.getMantisActiveQueryCountGauge().set((double) numSubs)); } subscriptionIdToStreamsMap.put(sub.getSubscriptionId(), streams); } StreamManager(Registry registry, MrePublishConfiguration mrePublishConfiguration); }
StreamManager { synchronized void addStreamSubscription(final Subscription sub) { List<String> streams = sanitizeStreamSubjects(sub.getSubjects()); LOG.info("adding subscription {} with streams {}", sub, streams); handleDuplicateSubscriptionId(sub); for (String stream : streams) { streamSubscriptionsMap.putIfAbsent(stream, new ConcurrentSkipListSet<>()); ConcurrentSkipListSet<Subscription> subs = streamSubscriptionsMap.get(stream); int maxSubs = config.maxSubscriptions(stream); subs.removeIf(s -> s.getSubscriptionId().equals(sub.getSubscriptionId())); subs.add(sub); int numSubs = subs.size(); if (numSubs > maxSubs) { removeStreamSubscription(sub); LOG.warn("QUERY SUBSCRIPTION REJECTED: Number of subscriptions for stream {} exceeded max {}. " + "Increase (default mre.publish.max.subscriptions.per.stream.default or " + " mre.publish.max.subscriptions.stream.<streamName>) to allow more queries. Removed {}", stream, maxSubs, sub); getStreamMetrics(stream).ifPresent(m -> m.getMantisQueryRejectedCounter().increment()); } getStreamMetrics(stream).ifPresent(m -> m.getMantisActiveQueryCountGauge().set((double) numSubs)); } subscriptionIdToStreamsMap.put(sub.getSubscriptionId(), streams); } StreamManager(Registry registry, MrePublishConfiguration mrePublishConfiguration);  }
StreamManager { synchronized void addStreamSubscription(final Subscription sub) { List<String> streams = sanitizeStreamSubjects(sub.getSubjects()); LOG.info("adding subscription {} with streams {}", sub, streams); handleDuplicateSubscriptionId(sub); for (String stream : streams) { streamSubscriptionsMap.putIfAbsent(stream, new ConcurrentSkipListSet<>()); ConcurrentSkipListSet<Subscription> subs = streamSubscriptionsMap.get(stream); int maxSubs = config.maxSubscriptions(stream); subs.removeIf(s -> s.getSubscriptionId().equals(sub.getSubscriptionId())); subs.add(sub); int numSubs = subs.size(); if (numSubs > maxSubs) { removeStreamSubscription(sub); LOG.warn("QUERY SUBSCRIPTION REJECTED: Number of subscriptions for stream {} exceeded max {}. " + "Increase (default mre.publish.max.subscriptions.per.stream.default or " + " mre.publish.max.subscriptions.stream.<streamName>) to allow more queries. Removed {}", stream, maxSubs, sub); getStreamMetrics(stream).ifPresent(m -> m.getMantisQueryRejectedCounter().increment()); } getStreamMetrics(stream).ifPresent(m -> m.getMantisActiveQueryCountGauge().set((double) numSubs)); } subscriptionIdToStreamsMap.put(sub.getSubscriptionId(), streams); } StreamManager(Registry registry, MrePublishConfiguration mrePublishConfiguration);   }