DbusEventBuffer implements Iterable<DbusEventInternalWritable>,
DbusEventBufferAppendable, DbusEventBufferStreamAppendable { @Override public void start(long startScn) { assert((_eventState == WindowState.INIT) || (_eventState == WindowState.ENDED)); startEvents(); endEvents(startScn); this.setPrevScn(startScn); } DbusEventBuffer(Config config);  DbusEventBuffer(StaticConfig config);  DbusEventBuffer(StaticConfig config, PhysicalPartition pPartition, DbusEventFactory eventFactory);  DbusEventBuffer(long maxEventBufferSize, int maxIndividualBufferSize, int maxIndexSize,
                         int initReadBufferSize, int maxEventSize,
                         AllocationPolicy allocationPolicy, File mmapDirectory,
                         QueuePolicy queuePolicy, RelayEventTraceOption traceOption,
                         PhysicalPartition physicalPartition, AssertLevel assertLevel, long bufRemovalWaitPeriod,
                         boolean restoreBuffers, boolean validateEventesInRestoredBuffers,
                         boolean enableScnIndex, DbusEventFactory eventFactory); static String getMmapMetaInfoFileNamePrefix(); static String getSessionPrefix(); void setDropOldEvents(boolean val); PhysicalPartition getPhysicalPartition(); synchronized void increaseRefCounter(); synchronized void decreaseRefCounter(); synchronized boolean shouldBeRemoved(boolean now); synchronized int getRefCount(); void forceReleaseDirectMemory(); void clear(); void reset(long prevScn); long getTimestampOfFirstEvent(); void validateEventsInBuffer(); static ByteBuffer allocateByteBuffer(int size, ByteOrder byteOrder,
                                       AllocationPolicy allocationPolicy,
                                       boolean restoreBuffers,
                                       File mmapSessionDir,
                                       File mmapFile); void setPrevScn(long scn); @Override long getPrevScn(); @Override long getMinScn(); @Override void start(long startScn); @Override void startEvents(); @Override boolean appendEvent(DbusEventKey key, short pPartitionId, short lPartitionId,
                             long timeStamp, short srcId, byte[] schemaId, byte[] value,
                             boolean enableTracing); @Override @Deprecated boolean appendEvent(
                             DbusEventKey key,
                             long sequenceId,
                             short pPartitionId,
                             short logicalPartitionId,
                             long timeStampInNanos,
                             short srcId,
                             byte[] schemaId,
                             byte[] value,
                             boolean enableTracing,
                             DbusEventsStatisticsCollector statsCollector); @Override boolean appendEvent(DbusEventKey key, short pPartitionId, short lPartitionId,
                             long timeStamp, short srcId, byte[] schemaId, byte[] value,
                             boolean enableTracing, DbusEventsStatisticsCollector statsCollector); @Override boolean appendEvent(DbusEventKey key, short pPartitionId, short lPartitionId,
                             long timeStamp, short srcId, byte[] schemaId, byte[] value,
                             boolean enableTracing, boolean isReplicated, DbusEventsStatisticsCollector statsCollector); @Override boolean appendEvent(DbusEventKey key,
                             DbusEventInfo eventInfo,
                             DbusEventsStatisticsCollector statsCollector); @Override void rollbackEvents(); @Override void endEvents(boolean updateWindowScn, long windowScn,
                        boolean updateIndex, boolean callListener,
                        DbusEventsStatisticsCollector statsCollector); void endEvents(long sequenceId); @Override void endEvents(long eventWindowScn,DbusEventsStatisticsCollector statsCollector); StreamEventsResult streamEvents(Checkpoint checkPoint,
                                         WritableByteChannel writeChannel,
                                         StreamEventsArgs args
                                         ); int batchWrite(Range range, WritableByteChannel writeChannel, Encoding encoding); @Override boolean empty(); int getReadStatus(); int readEvents(ReadableByteChannel readChannel, Encoding _encoding); int readEvents(ReadableByteChannel readChannel); int readEvents(ReadableByteChannel readChannel,
                        DbusEventsStatisticsCollector statsCollector); int readEvents(ReadableByteChannel readChannel,
                        Iterable<InternalDatabusEventsListener> eventListeners); @Override int readEvents(ReadableByteChannel readChannel,
                        Iterable<InternalDatabusEventsListener> eventListeners,
                        DbusEventsStatisticsCollector statsCollector); @Override String toString(); String bufPositionInfo(); String toShortString(); void waitForFreeSpaceUninterruptibly(long freeSpaceThreshold); void waitForFreeSpace(long freeSpaceThreshold); long getBufferFreeSpace(); int getBufferFreeReadSpace(); int getMaxReadBufferCapacity(); DbusEventIterator acquireIterator(String iteratorName); @Override Iterator<DbusEventInternalWritable> iterator(); void releaseIterator(BaseEventIterator e); void addInternalListener(InternalDatabusEventsListener listener); boolean removeInternalListener(InternalDatabusEventsListener listener); BufferPositionParser getBufferPositionParser(); long getAllocatedSize(); @Override long lastWrittenScn(); @Override void setStartSCN(long sinceSCN); long getTimestampOfLatestDataEvent(); void closeBuffer(boolean persistBuffer); void removeMMapFiles(); void cleanUpPersistedBuffers(); void saveBufferMetaInfo(boolean infoOnly); void initBuffersWithMetaInfo(DbusEventBufferMetaInfo mi); long getSeenEndOfPeriodScn(); boolean isSCNRegress(); Logger getLog(); String hexdumpByteBufferContents(long pos, int length); boolean injectEvent(DbusEventInternalReadable event); synchronized byte getEventSerializationVersion(); static final String MODULE; static final Logger LOG; static final String MMAP_META_INFO_FILE_NAME; static final String SESSION_PREFIX; static final String MMAP_META_INFO_SUFFIX; static final String PERF_MODULE; static final Logger PERF_LOG; static final int MAX_DEBUG_ON_ERROR_ITERATIONS; }
@Test public void testSessionIdGeneration() throws Exception { class IdGeneratorThread implements Runnable { private final DbusEventBuffer.SessionIdGenerator _idGenerator; private final int _nTimes; private final String _ids[]; public IdGeneratorThread(DbusEventBuffer.SessionIdGenerator idGenerator, int nTimes) { _idGenerator = idGenerator; _nTimes = nTimes; _ids = new String[_nTimes]; } @Override public void run() { for (int i = 0; i < _nTimes; i++) { _ids[i] = _idGenerator.generateSessionId(); } for (int i = 0; i < _nTimes; i++) { try { addId(_ids[i]); } catch (Exception e) { throw new RuntimeException("Could not add Id" + _ids[i]); } } thrCounter.decrementAndGet(); } } final int nThreads = 20; final int nTimes = 300; final Thread threads[] = new Thread[nThreads]; final DbusEventBuffer.SessionIdGenerator idGenerator = new DbusEventBuffer.SessionIdGenerator(); for (int i = 0; i < nThreads; i++) { threads[i] = new Thread(new IdGeneratorThread(idGenerator, nTimes), "idgen_thread_" + i); } thrCounter.set(nThreads); for (int i = 0; i < nThreads; i++) { threads[i].start(); } for (int i = 0; i < nThreads; i++) { threads[i].join(); } Assert.assertEquals(0, thrCounter.get()); }