DbusEventBufferMult { public DbusFilter constructFilters(Collection<DatabusSubscription> subs) throws DatabusException { HashMap<PhysicalPartition, PhysicalPartitionDbusFilter> filterMap = null; for (DatabusSubscription sub: subs) { PhysicalPartition ppart = sub.getPhysicalPartition(); if (sub.getLogicalSource().isWildcard()) { if (!ppart.isWildcard()) { if (null == filterMap) filterMap = new HashMap<PhysicalPartition, PhysicalPartitionDbusFilter>(10); filterMap.put(ppart, new PhysicalPartitionDbusFilter(ppart, null)); } else { LOG.warn("ignoring subscription with both physical partition and logical source wildcards"); } } else { PhysicalPartitionDbusFilter ppartFilter = null != filterMap ? filterMap.get(ppart) : null; LogicalSourceAndPartitionDbusFilter logFilter = null; if (null == ppartFilter) { logFilter = new LogicalSourceAndPartitionDbusFilter(); ppartFilter = new PhysicalPartitionDbusFilter(ppart, logFilter); if (null == filterMap) filterMap = new HashMap<PhysicalPartition, PhysicalPartitionDbusFilter>(10); filterMap.put(ppart, ppartFilter); } else { logFilter = (LogicalSourceAndPartitionDbusFilter)ppartFilter.getNestedFilter(); } if (null != logFilter) logFilter.addSourceCondition(sub.getLogicalPartition()); else LOG.error("unexpected null filter for logical source"); } } if (0 == filterMap.size()) return AllowAllDbusFilter.THE_INSTANCE; else if (1 == filterMap.size()) { DbusFilter result = filterMap.entrySet().iterator().next().getValue(); return result; } else { ConjunctionDbusFilter result = new ConjunctionDbusFilter(); for (Map.Entry<PhysicalPartition, PhysicalPartitionDbusFilter> filterEntry: filterMap.entrySet()) { result.addFilter(filterEntry.getValue()); } return result; } } DbusEventBufferMult();  DbusEventBufferMult(PhysicalSourceStaticConfig [] pConfigs,
                             DbusEventBuffer.StaticConfig config,
                             DbusEventFactory eventFactory); DbusEventBuffer getOneBuffer(LogicalSource lSource,
                                       LogicalPartition lPartition); DbusEventBuffer getOneBuffer( PhysicalPartition pPartition); void resetBuffer(PhysicalPartition pPartition, long prevScn); DbusEventBufferAppendable getDbusEventBufferAppendable(PhysicalPartition pPartition); DbusEventBufferAppendable getDbusEventBufferAppendable(int lSrcId); DbusEventBufferAppendable getDbusEventBufferAppendable(LogicalSource lSource); DbusEventBufferAppendable getDbusEventBufferAppendable(LogicalSource lSource,
                                                                LogicalPartition lPartition); DbusEventBuffer getDbusEventBuffer(LogicalSource lSource); DbusEventBufferBatchReadable getDbusEventBufferBatchReadable(CheckpointMult cpMult,
            Set<PhysicalPartitionKey> ppartKeys, StatsCollectors<DbusEventsStatisticsCollector> statsCollectors); DbusEventBufferBatchReadable getDbusEventBufferBatchReadable(Collection<Integer> ids,
		  CheckpointMult cpMult, StatsCollectors<DbusEventsStatisticsCollector> statsCollector); DbusEventBufferBatchReadable getDbusEventBufferBatchReadable(
         CheckpointMult cpMult,
         Collection<PhysicalPartitionKey> physicalPartitions,
         StatsCollectors<DbusEventsStatisticsCollector> statsCollector); PhysicalPartition getPhysicalPartition(int srcId ); PhysicalPartition getPhysicalPartition(int srcId, LogicalPartition lPartition ); Iterable<DbusEventBuffer> bufIterable(); synchronized DbusEventBuffer addNewBuffer(PhysicalSourceStaticConfig pConfig,
                                                   DbusEventBuffer.StaticConfig config); synchronized void removeBuffer(PhysicalPartitionKey pKey, PhysicalSource pSource); synchronized void removeBuffer(PhysicalSourceStaticConfig pConfig); synchronized void deallocateRemovedBuffers(boolean now); synchronized void close(); void assertBuffers(); synchronized void addBuffer(PhysicalSourceStaticConfig pConfig, DbusEventBuffer buf); Set<PhysicalSource> getPhysicalSourcesForPartition(PhysicalPartition pPart); DbusFilter constructFilters(Collection<DatabusSubscription> subs); NavigableSet<PhysicalPartitionKey> getAllPhysicalPartitionKeys(); void setDropOldEvents(boolean val); int bufsNum(); void startAllEvents(); void endAllEvents(long seq, long nTime, DbusEventsStatisticsCollector stats); void clearAll(); synchronized void saveBufferMetaInfo(boolean infoOnly); void validateRelayBuffers(); void rollbackAllBuffers(); static final String MODULE; static final Logger LOG; static final String PERF_MODULE; static final Logger PERF_LOG; static final String BAK_DIRNAME_SUFFIX; }
@Test public void testConstructFilters() throws Exception { TestSetup t = new TestSetup(); DatabusSubscription sub1 = DatabusSubscription.createPhysicalPartitionReplicationSubscription(new PhysicalPartition(100, "multBufferTest1")); DbusFilter filter1 = t._eventBuffer.constructFilters(Arrays.asList(sub1)); assertNotNull(filter1); assertTrue(filter1 instanceof PhysicalPartitionDbusFilter); PhysicalPartitionDbusFilter ppfilter1 = (PhysicalPartitionDbusFilter)filter1; assertEquals(ppfilter1.getPhysicalPartition(), new PhysicalPartition(100, "multBufferTest1")); assertNull(ppfilter1.getNestedFilter()); DatabusSubscription sub2 = DatabusSubscription.createPhysicalPartitionReplicationSubscription(new PhysicalPartition(101, "multBufferTest2")); DbusFilter filter2 = t._eventBuffer.constructFilters(Arrays.asList(sub1, sub2)); assertNotNull(filter2); assertTrue(filter2 instanceof ConjunctionDbusFilter); ConjunctionDbusFilter conjFilter2 = (ConjunctionDbusFilter)filter2; boolean hasPP100 = false; boolean hasPP101 = false; assertEquals(conjFilter2.getFilterList().size(), 2); for (DbusFilter f: conjFilter2.getFilterList()) { assertTrue(f instanceof PhysicalPartitionDbusFilter); PhysicalPartitionDbusFilter ppf = (PhysicalPartitionDbusFilter)f; if (ppf.getPhysicalPartition().getId() == 100) hasPP100 = true; else if (ppf.getPhysicalPartition().getId() == 101) hasPP101 = true; else fail("unknown physical partition filter:" + ppf.getPhysicalPartition()); } assertTrue(hasPP100); assertTrue(hasPP101); DatabusSubscription sub3 = DatabusSubscription.createSimpleSourceSubscription(new LogicalSource(2, "srcName2")); DbusFilter filter3 = t._eventBuffer.constructFilters(Arrays.asList(sub3)); assertNotNull(filter3); assertTrue(filter3 instanceof PhysicalPartitionDbusFilter); PhysicalPartitionDbusFilter ppfilter3 = (PhysicalPartitionDbusFilter)filter3; assertEquals(ppfilter3.getPhysicalPartition(), PhysicalPartition.ANY_PHYSICAL_PARTITION); DbusFilter ppfilter3_child = ppfilter3.getNestedFilter(); assertNotNull(ppfilter3_child); assertTrue(ppfilter3_child instanceof LogicalSourceAndPartitionDbusFilter); LogicalSourceAndPartitionDbusFilter lsourceFilter3 = (LogicalSourceAndPartitionDbusFilter)ppfilter3_child; LogicalSourceAndPartitionDbusFilter.LogicalPartitionDbusFilter lpartFilter3_1 = lsourceFilter3.getSourceFilter(2); assertNotNull(lpartFilter3_1); assertTrue(lpartFilter3_1.isAllPartitionsWildcard()); DatabusSubscription sub4 = new DatabusSubscription(PhysicalSource.MASTER_PHISYCAL_SOURCE, new PhysicalPartition(101, "multBufferTest2"), new LogicalSourceId(new LogicalSource(2, "srcName2"), (short)2) ); DbusFilter filter4 = t._eventBuffer.constructFilters(Arrays.asList(sub4)); assertNotNull(filter4); assertTrue(filter4 instanceof PhysicalPartitionDbusFilter); PhysicalPartitionDbusFilter ppfilter4 = (PhysicalPartitionDbusFilter)filter4; assertEquals(ppfilter4.getPhysicalPartition(), new PhysicalPartition(101, "multBufferTest2")); DbusFilter ppfilter4_child = ppfilter4.getNestedFilter(); assertNotNull(ppfilter4_child); assertTrue(ppfilter4_child instanceof LogicalSourceAndPartitionDbusFilter); LogicalSourceAndPartitionDbusFilter lsourceFilter4 = (LogicalSourceAndPartitionDbusFilter)ppfilter4_child; LogicalSourceAndPartitionDbusFilter.LogicalPartitionDbusFilter lpartFilter4_1 = lsourceFilter4.getSourceFilter(2); assertNotNull(lpartFilter4_1); assertTrue(lpartFilter4_1.getPartitionsMask().contains(2)); }