@Test public void testEmitEdges() { final int version = 0; final ConnectionService connectionService = mock( ConnectionService.class ); final AllApplicationsObservable allApplicationsObservable = mock( AllApplicationsObservable.class ); final int count = 3000; final Observable<ConnectionScope> edgeEmitter = Observable.create( new DupConnectionEmitter( count ) ); when( connectionService.deDupeConnections( any( Observable.class ) ) ).thenReturn( edgeEmitter ); final DeDupConnectionDataMigration plugin = new DeDupConnectionDataMigration( connectionService, allApplicationsObservable ); final TestProgressObserver testProgressObserver = new TestProgressObserver(); plugin.migrate( version, testProgressObserver ); final List<String> updates = testProgressObserver.getUpdates(); assertEquals( "Expected 3 updates", 3, updates.size() ); assertEquals( "De duped 1000 edges", updates.get( 0 ) ); assertEquals( "De duped 2000 edges", updates.get( 1 ) ); assertEquals( "De duped 3000 edges", updates.get( 2 ) ); assertTrue( "Should complete", testProgressObserver.isComplete() ); assertTrue( "Should start", testProgressObserver.isStarted() ); assertFalse( "Should not fail", testProgressObserver.isFailed() ); }
@Override public int migrate( final int currentVersion, final ProgressObserver observer ) { final int migrationVersion = getMaxVersion(); observer.start(); connectionService.deDupeConnections( allApplicationsObservable.getData() ).reduce( 0l, ( count, deDuped ) -> { final long newCount = count + 1; if ( newCount % UPDATE_COUNT == 0 ) { logger.info( "De duped {} edges", newCount ); observer.update( migrationVersion, String.format( "De duped %d edges", newCount ) ); } return newCount; } ).doOnNext( total -> { logger.info( "Completed de-duping {} edges", total ); observer.complete(); } ).toBlocking().lastOrDefault( null ); return migrationVersion; }
DeDupConnectionDataMigration implements DataMigration { @Override public int migrate( final int currentVersion, final ProgressObserver observer ) { final int migrationVersion = getMaxVersion(); observer.start(); connectionService.deDupeConnections( allApplicationsObservable.getData() ).reduce( 0l, ( count, deDuped ) -> { final long newCount = count + 1; if ( newCount % UPDATE_COUNT == 0 ) { logger.info( "De duped {} edges", newCount ); observer.update( migrationVersion, String.format( "De duped %d edges", newCount ) ); } return newCount; } ).doOnNext( total -> { logger.info( "Completed de-duping {} edges", total ); observer.complete(); } ).toBlocking().lastOrDefault( null ); return migrationVersion; } }
DeDupConnectionDataMigration implements DataMigration { @Override public int migrate( final int currentVersion, final ProgressObserver observer ) { final int migrationVersion = getMaxVersion(); observer.start(); connectionService.deDupeConnections( allApplicationsObservable.getData() ).reduce( 0l, ( count, deDuped ) -> { final long newCount = count + 1; if ( newCount % UPDATE_COUNT == 0 ) { logger.info( "De duped {} edges", newCount ); observer.update( migrationVersion, String.format( "De duped %d edges", newCount ) ); } return newCount; } ).doOnNext( total -> { logger.info( "Completed de-duping {} edges", total ); observer.complete(); } ).toBlocking().lastOrDefault( null ); return migrationVersion; } @Inject  DeDupConnectionDataMigration( final ConnectionService connectionService,
                                         final AllApplicationsObservable allApplicationsObservable ); }
DeDupConnectionDataMigration implements DataMigration { @Override public int migrate( final int currentVersion, final ProgressObserver observer ) { final int migrationVersion = getMaxVersion(); observer.start(); connectionService.deDupeConnections( allApplicationsObservable.getData() ).reduce( 0l, ( count, deDuped ) -> { final long newCount = count + 1; if ( newCount % UPDATE_COUNT == 0 ) { logger.info( "De duped {} edges", newCount ); observer.update( migrationVersion, String.format( "De duped %d edges", newCount ) ); } return newCount; } ).doOnNext( total -> { logger.info( "Completed de-duping {} edges", total ); observer.complete(); } ).toBlocking().lastOrDefault( null ); return migrationVersion; } @Inject  DeDupConnectionDataMigration( final ConnectionService connectionService,
                                         final AllApplicationsObservable allApplicationsObservable ); @Override int migrate( final int currentVersion, final ProgressObserver observer ); @Override boolean supports( final int currentVersion ); @Override int getMaxVersion(); }
DeDupConnectionDataMigration implements DataMigration { @Override public int migrate( final int currentVersion, final ProgressObserver observer ) { final int migrationVersion = getMaxVersion(); observer.start(); connectionService.deDupeConnections( allApplicationsObservable.getData() ).reduce( 0l, ( count, deDuped ) -> { final long newCount = count + 1; if ( newCount % UPDATE_COUNT == 0 ) { logger.info( "De duped {} edges", newCount ); observer.update( migrationVersion, String.format( "De duped %d edges", newCount ) ); } return newCount; } ).doOnNext( total -> { logger.info( "Completed de-duping {} edges", total ); observer.complete(); } ).toBlocking().lastOrDefault( null ); return migrationVersion; } @Inject  DeDupConnectionDataMigration( final ConnectionService connectionService,
                                         final AllApplicationsObservable allApplicationsObservable ); @Override int migrate( final int currentVersion, final ProgressObserver observer ); @Override boolean supports( final int currentVersion ); @Override int getMaxVersion();  }