@Test public void filter() throws JoyQueueException { List<ByteBuffer> byteBufferList = new LinkedList<>(); for (int i = 0; i < 10; i++) { ByteBuffer allocate = ByteBuffer.allocate(100); allocate.position(59); allocate.putShort((short) i); allocate.flip(); byteBufferList.add(allocate); } List<ByteBuffer> filter = filterMessageSupport.filter(consumer, byteBufferList, new FilterCallback() { @Override public void callback(List<ByteBuffer> list) throws JoyQueueException { Assert.assertEquals(1, list.size()); } }); Assert.assertEquals(3, filter.size()); }
public List<ByteBuffer> filter(Consumer consumer, List<ByteBuffer> byteBuffers, FilterCallback filterCallback) throws JoyQueueException { FilterPipeline<MessageFilter> filterPipeline = filterRuleCache.get(consumer.getId()); if (filterPipeline == null) { filterPipeline = createFilterPipeline(consumer.getConsumerPolicy()); filterRuleCache.putIfAbsent(consumer.getId(), filterPipeline); } List<ByteBuffer> result = filterPipeline.execute(byteBuffers, filterCallback); return result; }
FilterMessageSupport { public List<ByteBuffer> filter(Consumer consumer, List<ByteBuffer> byteBuffers, FilterCallback filterCallback) throws JoyQueueException { FilterPipeline<MessageFilter> filterPipeline = filterRuleCache.get(consumer.getId()); if (filterPipeline == null) { filterPipeline = createFilterPipeline(consumer.getConsumerPolicy()); filterRuleCache.putIfAbsent(consumer.getId(), filterPipeline); } List<ByteBuffer> result = filterPipeline.execute(byteBuffers, filterCallback); return result; } }
FilterMessageSupport { public List<ByteBuffer> filter(Consumer consumer, List<ByteBuffer> byteBuffers, FilterCallback filterCallback) throws JoyQueueException { FilterPipeline<MessageFilter> filterPipeline = filterRuleCache.get(consumer.getId()); if (filterPipeline == null) { filterPipeline = createFilterPipeline(consumer.getConsumerPolicy()); filterRuleCache.putIfAbsent(consumer.getId(), filterPipeline); } List<ByteBuffer> result = filterPipeline.execute(byteBuffers, filterCallback); return result; } FilterMessageSupport(ClusterManager clusterManager); }
FilterMessageSupport { public List<ByteBuffer> filter(Consumer consumer, List<ByteBuffer> byteBuffers, FilterCallback filterCallback) throws JoyQueueException { FilterPipeline<MessageFilter> filterPipeline = filterRuleCache.get(consumer.getId()); if (filterPipeline == null) { filterPipeline = createFilterPipeline(consumer.getConsumerPolicy()); filterRuleCache.putIfAbsent(consumer.getId(), filterPipeline); } List<ByteBuffer> result = filterPipeline.execute(byteBuffers, filterCallback); return result; } FilterMessageSupport(ClusterManager clusterManager); List<ByteBuffer> filter(Consumer consumer, List<ByteBuffer> byteBuffers, FilterCallback filterCallback); }
FilterMessageSupport { public List<ByteBuffer> filter(Consumer consumer, List<ByteBuffer> byteBuffers, FilterCallback filterCallback) throws JoyQueueException { FilterPipeline<MessageFilter> filterPipeline = filterRuleCache.get(consumer.getId()); if (filterPipeline == null) { filterPipeline = createFilterPipeline(consumer.getConsumerPolicy()); filterRuleCache.putIfAbsent(consumer.getId(), filterPipeline); } List<ByteBuffer> result = filterPipeline.execute(byteBuffers, filterCallback); return result; } FilterMessageSupport(ClusterManager clusterManager); List<ByteBuffer> filter(Consumer consumer, List<ByteBuffer> byteBuffers, FilterCallback filterCallback);  }