public void addSubscriber ( Handler handler ) { String handlerId = CollectorManagerUtils . getHandlerId ( handler ) ; //temporary exception for audit Source //must change later if ( source . getSourceName ( ) . trim ( ) . equals ( "audit" ) && subscribers . isEmpty ( ) ) { //First subscriber, assign a buffer. bufferMgr = new BufferManagerImpl ( BUFFER_SIZE , sourceId , false ) ; //Inform the source that a buffer is now available //and it can start sending events to this buffer. source . setBufferManager ( this . bufferMgr ) ; } subscribers . add ( handlerId ) ; /* * Inform the handler that this source/buffer/conduit is now available: * Synchronous Handler - Add handler as synchronous handler to the Buffer/Conduit's sync handler set * Asynchronous Handler - Add handler as asynchronous handler into the Buffer/Conduit * */ if ( handler instanceof SynchronousHandler ) { bufferMgr . addSyncHandler ( ( SynchronousHandler ) handler ) ; } else { bufferMgr . addHandler ( handlerId ) ; } }
Method to add a subscriber to the source Source manager will allocate a buffer when the source has atleast one subscriber .