@Test public void testDuplicateConnections() { final ApplicationScope applicationScope = new ApplicationScopeImpl( new SimpleId( "application" ) ); final GraphManager gm = graphManagerFactory.createEdgeManager( applicationScope ); final Id source = new SimpleId( "source" ); final String collectionName = "testCollection"; final Edge collectionEdge = CpNamingUtils.createCollectionEdge( applicationScope.getApplication(), collectionName, source ); final Edge writtenCollection = gm.writeEdge( collectionEdge ).toBlocking().last(); assertNotNull("Collection edge written", writtenCollection); final Id target = new SimpleId( "target" ); final String connectionType = "testConnection"; final Edge connection1 = CpNamingUtils.createConnectionEdge( source, connectionType, target ); final Edge written1 = gm.writeEdge( connection1 ).toBlocking().last(); logger.info( "Wrote edge 1 with edge {}", written1 ); final Edge connection2 = CpNamingUtils.createConnectionEdge( source, connectionType, target ); final Edge written2 = gm.writeEdge( connection2 ).toBlocking().last(); logger.info( "Wrote edge 2 with edge {}", written2 ); final Edge connection3 = CpNamingUtils.createConnectionEdge( source, connectionType, target ); final Edge written3 = gm.writeEdge( connection3 ).toBlocking().last(); logger.info( "Wrote edge 3 with edge {}", written3 ); assertTrue( "Expected edge timestamp to be in order", written1.getTimestamp() <= written2.getTimestamp() ); assertTrue( "Expected edge timestamp to be in order", written2.getTimestamp() <= written3.getTimestamp() ); final List<ConnectionScope> deletedConnections = connectionService.deDupeConnections( Observable.just( applicationScope ) ).toList().toBlocking().last(); assertEdgeData( written2, deletedConnections.get( 0 ).getEdge() ); assertEdgeData( written1, deletedConnections.get( 1 ).getEdge() ); assertEquals( "2 edges deleted", 2, deletedConnections.size() ); final SearchByEdge simpleSearchByEdge = new SimpleSearchByEdge( source, connection1.getType(), target, Long.MAX_VALUE, SearchByEdgeType.Order.DESCENDING, Optional.absent() ); final List<MarkedEdge> edges = gm.loadEdgeVersions( simpleSearchByEdge ).toList().toBlocking().last(); assertEquals( 1, edges.size() ); assertEquals( written3, edges.get( 0 ) ); }
@Override public Observable<ConnectionScope> deDupeConnections( final Observable<ApplicationScope> applicationScopeObservable ) { final Observable<EntityIdScope> entityIds = allEntityIdsObservable.getEntities( applicationScopeObservable ); return entityIds.flatMap( entityIdScope -> { final ApplicationScope applicationScope = entityIdScope.getApplicationScope(); final Id entityId = entityIdScope.getId(); final GraphManager gm = graphManagerFactory.createEdgeManager( applicationScope ); if (logger.isDebugEnabled()) { logger.debug("Checking connections of id {} in application {}", entityId, applicationScope); } return gm.getEdgeTypesFromSource( new SimpleSearchEdgeType( entityId, CpNamingUtils.EDGE_CONN_PREFIX, Optional.absent() ) ) .flatMap( edgeType -> { if (logger.isDebugEnabled()) { logger.debug("Found edge of types of {}, searching for edges", edgeType); } final SearchByEdgeType searchByEdge = new SimpleSearchByEdgeType( entityId, edgeType, Long.MAX_VALUE, SearchByEdgeType.Order.DESCENDING, Optional.absent() ); return gm.loadEdgesFromSource( searchByEdge ); } ) .flatMap( edge -> { if (logger.isDebugEnabled()) { logger.debug("Found edge {}, searching for multiple versions of edge", edge); } final SearchByEdge searchByEdge = new SimpleSearchByEdge( edge.getSourceNode(), edge.getType(), edge.getTargetNode(), Long.MAX_VALUE, SearchByEdgeType.Order.DESCENDING, Optional.absent() ); return gm.loadEdgeVersions( searchByEdge ) .skip( 1 ) .flatMap( edgeToDelete -> { if (logger.isDebugEnabled()) { logger.debug("Deleting edge {}", edgeToDelete); } return gm.markEdge( edgeToDelete ); } ) .flatMap( lastMarkedEdge -> gm.deleteEdge( lastMarkedEdge ) ); } ).map( deletedEdge -> new ConnectionScope( applicationScope, deletedEdge ) ); } ); }
ConnectionServiceImpl implements ConnectionService { @Override public Observable<ConnectionScope> deDupeConnections( final Observable<ApplicationScope> applicationScopeObservable ) { final Observable<EntityIdScope> entityIds = allEntityIdsObservable.getEntities( applicationScopeObservable ); return entityIds.flatMap( entityIdScope -> { final ApplicationScope applicationScope = entityIdScope.getApplicationScope(); final Id entityId = entityIdScope.getId(); final GraphManager gm = graphManagerFactory.createEdgeManager( applicationScope ); if (logger.isDebugEnabled()) { logger.debug("Checking connections of id {} in application {}", entityId, applicationScope); } return gm.getEdgeTypesFromSource( new SimpleSearchEdgeType( entityId, CpNamingUtils.EDGE_CONN_PREFIX, Optional.absent() ) ) .flatMap( edgeType -> { if (logger.isDebugEnabled()) { logger.debug("Found edge of types of {}, searching for edges", edgeType); } final SearchByEdgeType searchByEdge = new SimpleSearchByEdgeType( entityId, edgeType, Long.MAX_VALUE, SearchByEdgeType.Order.DESCENDING, Optional.absent() ); return gm.loadEdgesFromSource( searchByEdge ); } ) .flatMap( edge -> { if (logger.isDebugEnabled()) { logger.debug("Found edge {}, searching for multiple versions of edge", edge); } final SearchByEdge searchByEdge = new SimpleSearchByEdge( edge.getSourceNode(), edge.getType(), edge.getTargetNode(), Long.MAX_VALUE, SearchByEdgeType.Order.DESCENDING, Optional.absent() ); return gm.loadEdgeVersions( searchByEdge ) .skip( 1 ) .flatMap( edgeToDelete -> { if (logger.isDebugEnabled()) { logger.debug("Deleting edge {}", edgeToDelete); } return gm.markEdge( edgeToDelete ); } ) .flatMap( lastMarkedEdge -> gm.deleteEdge( lastMarkedEdge ) ); } ).map( deletedEdge -> new ConnectionScope( applicationScope, deletedEdge ) ); } ); } }
ConnectionServiceImpl implements ConnectionService { @Override public Observable<ConnectionScope> deDupeConnections( final Observable<ApplicationScope> applicationScopeObservable ) { final Observable<EntityIdScope> entityIds = allEntityIdsObservable.getEntities( applicationScopeObservable ); return entityIds.flatMap( entityIdScope -> { final ApplicationScope applicationScope = entityIdScope.getApplicationScope(); final Id entityId = entityIdScope.getId(); final GraphManager gm = graphManagerFactory.createEdgeManager( applicationScope ); if (logger.isDebugEnabled()) { logger.debug("Checking connections of id {} in application {}", entityId, applicationScope); } return gm.getEdgeTypesFromSource( new SimpleSearchEdgeType( entityId, CpNamingUtils.EDGE_CONN_PREFIX, Optional.absent() ) ) .flatMap( edgeType -> { if (logger.isDebugEnabled()) { logger.debug("Found edge of types of {}, searching for edges", edgeType); } final SearchByEdgeType searchByEdge = new SimpleSearchByEdgeType( entityId, edgeType, Long.MAX_VALUE, SearchByEdgeType.Order.DESCENDING, Optional.absent() ); return gm.loadEdgesFromSource( searchByEdge ); } ) .flatMap( edge -> { if (logger.isDebugEnabled()) { logger.debug("Found edge {}, searching for multiple versions of edge", edge); } final SearchByEdge searchByEdge = new SimpleSearchByEdge( edge.getSourceNode(), edge.getType(), edge.getTargetNode(), Long.MAX_VALUE, SearchByEdgeType.Order.DESCENDING, Optional.absent() ); return gm.loadEdgeVersions( searchByEdge ) .skip( 1 ) .flatMap( edgeToDelete -> { if (logger.isDebugEnabled()) { logger.debug("Deleting edge {}", edgeToDelete); } return gm.markEdge( edgeToDelete ); } ) .flatMap( lastMarkedEdge -> gm.deleteEdge( lastMarkedEdge ) ); } ).map( deletedEdge -> new ConnectionScope( applicationScope, deletedEdge ) ); } ); } @Inject  ConnectionServiceImpl( final PipelineBuilderFactory pipelineBuilderFactory,
                                  final AllEntityIdsObservable allEntityIdsObservable,
                                  final GraphManagerFactory graphManagerFactory ); }
ConnectionServiceImpl implements ConnectionService { @Override public Observable<ConnectionScope> deDupeConnections( final Observable<ApplicationScope> applicationScopeObservable ) { final Observable<EntityIdScope> entityIds = allEntityIdsObservable.getEntities( applicationScopeObservable ); return entityIds.flatMap( entityIdScope -> { final ApplicationScope applicationScope = entityIdScope.getApplicationScope(); final Id entityId = entityIdScope.getId(); final GraphManager gm = graphManagerFactory.createEdgeManager( applicationScope ); if (logger.isDebugEnabled()) { logger.debug("Checking connections of id {} in application {}", entityId, applicationScope); } return gm.getEdgeTypesFromSource( new SimpleSearchEdgeType( entityId, CpNamingUtils.EDGE_CONN_PREFIX, Optional.absent() ) ) .flatMap( edgeType -> { if (logger.isDebugEnabled()) { logger.debug("Found edge of types of {}, searching for edges", edgeType); } final SearchByEdgeType searchByEdge = new SimpleSearchByEdgeType( entityId, edgeType, Long.MAX_VALUE, SearchByEdgeType.Order.DESCENDING, Optional.absent() ); return gm.loadEdgesFromSource( searchByEdge ); } ) .flatMap( edge -> { if (logger.isDebugEnabled()) { logger.debug("Found edge {}, searching for multiple versions of edge", edge); } final SearchByEdge searchByEdge = new SimpleSearchByEdge( edge.getSourceNode(), edge.getType(), edge.getTargetNode(), Long.MAX_VALUE, SearchByEdgeType.Order.DESCENDING, Optional.absent() ); return gm.loadEdgeVersions( searchByEdge ) .skip( 1 ) .flatMap( edgeToDelete -> { if (logger.isDebugEnabled()) { logger.debug("Deleting edge {}", edgeToDelete); } return gm.markEdge( edgeToDelete ); } ) .flatMap( lastMarkedEdge -> gm.deleteEdge( lastMarkedEdge ) ); } ).map( deletedEdge -> new ConnectionScope( applicationScope, deletedEdge ) ); } ); } @Inject  ConnectionServiceImpl( final PipelineBuilderFactory pipelineBuilderFactory,
                                  final AllEntityIdsObservable allEntityIdsObservable,
                                  final GraphManagerFactory graphManagerFactory ); @Override Observable<ResultsPage<Entity>> searchConnection( final ConnectionSearch search ); @Override Observable<ResultsPage<ConnectionRef>> searchConnectionAsRefs( final ConnectionSearch search ); @Override Observable<ConnectionScope> deDupeConnections(
        final Observable<ApplicationScope> applicationScopeObservable ); }
ConnectionServiceImpl implements ConnectionService { @Override public Observable<ConnectionScope> deDupeConnections( final Observable<ApplicationScope> applicationScopeObservable ) { final Observable<EntityIdScope> entityIds = allEntityIdsObservable.getEntities( applicationScopeObservable ); return entityIds.flatMap( entityIdScope -> { final ApplicationScope applicationScope = entityIdScope.getApplicationScope(); final Id entityId = entityIdScope.getId(); final GraphManager gm = graphManagerFactory.createEdgeManager( applicationScope ); if (logger.isDebugEnabled()) { logger.debug("Checking connections of id {} in application {}", entityId, applicationScope); } return gm.getEdgeTypesFromSource( new SimpleSearchEdgeType( entityId, CpNamingUtils.EDGE_CONN_PREFIX, Optional.absent() ) ) .flatMap( edgeType -> { if (logger.isDebugEnabled()) { logger.debug("Found edge of types of {}, searching for edges", edgeType); } final SearchByEdgeType searchByEdge = new SimpleSearchByEdgeType( entityId, edgeType, Long.MAX_VALUE, SearchByEdgeType.Order.DESCENDING, Optional.absent() ); return gm.loadEdgesFromSource( searchByEdge ); } ) .flatMap( edge -> { if (logger.isDebugEnabled()) { logger.debug("Found edge {}, searching for multiple versions of edge", edge); } final SearchByEdge searchByEdge = new SimpleSearchByEdge( edge.getSourceNode(), edge.getType(), edge.getTargetNode(), Long.MAX_VALUE, SearchByEdgeType.Order.DESCENDING, Optional.absent() ); return gm.loadEdgeVersions( searchByEdge ) .skip( 1 ) .flatMap( edgeToDelete -> { if (logger.isDebugEnabled()) { logger.debug("Deleting edge {}", edgeToDelete); } return gm.markEdge( edgeToDelete ); } ) .flatMap( lastMarkedEdge -> gm.deleteEdge( lastMarkedEdge ) ); } ).map( deletedEdge -> new ConnectionScope( applicationScope, deletedEdge ) ); } ); } @Inject  ConnectionServiceImpl( final PipelineBuilderFactory pipelineBuilderFactory,
                                  final AllEntityIdsObservable allEntityIdsObservable,
                                  final GraphManagerFactory graphManagerFactory ); @Override Observable<ResultsPage<Entity>> searchConnection( final ConnectionSearch search ); @Override Observable<ResultsPage<ConnectionRef>> searchConnectionAsRefs( final ConnectionSearch search ); @Override Observable<ConnectionScope> deDupeConnections(
        final Observable<ApplicationScope> applicationScopeObservable );  }