boolean openStream ( ) throws InterruptedException , IOException { logger . info ( "stream START" ) ; final boolean isOpen ; final Set < BsonValue > idsToWatch = nsConfig . getSynchronizedDocumentIds ( ) ; if ( ! networkMonitor . isConnected ( ) ) { logger . info ( "stream END - Network disconnected" ) ; return false ; } if ( idsToWatch . isEmpty ( ) ) { logger . info ( "stream END - No synchronized documents" ) ; return false ; } nsLock . writeLock ( ) . lockInterruptibly ( ) ; try { if ( ! authMonitor . isLoggedIn ( ) ) { logger . info ( "stream END - Logged out" ) ; return false ; } final Document args = new Document ( ) ; args . put ( "database" , namespace . getDatabaseName ( ) ) ; args . put ( "collection" , namespace . getCollectionName ( ) ) ; args . put ( "ids" , idsToWatch ) ; currentStream = service . streamFunction ( "watch" , Collections . singletonList ( args ) , ResultDecoders . changeEventDecoder ( BSON_DOCUMENT_CODEC ) ) ; if ( currentStream != null && currentStream . isOpen ( ) ) { this . nsConfig . setStale ( true ) ; isOpen = true ; } else { isOpen = false ; } } finally { nsLock . writeLock ( ) . unlock ( ) ; } return isOpen ; }
Open the event stream