@Test(expected = NullPointerException.class) public void loadParamVersion() throws ConnectionException { serializationStrategy .load( new ApplicationScopeImpl(new SimpleId( "organization" ) ), Collections.<Id>singleton( new SimpleId( "test" )), null ); }
@Override public EntitySet load( final ApplicationScope applicationScope, final Collection<Id> entityIds, final UUID maxVersion ) { Preconditions.checkNotNull( applicationScope, "applicationScope is required" ); Preconditions.checkNotNull( entityIds, "entityIds is required" ); Preconditions.checkArgument( entityIds.size() > 0, "entityIds is required" ); Preconditions.checkNotNull( maxVersion, "version is required" ); if( entityIds.size() > serializationFig.getMaxLoadSize()){ throw new IllegalArgumentException( "requested load size cannot be over configured maximum of " + serializationFig.getMaxLoadSize() ); } final Id applicationId = applicationScope.getApplication(); final Id ownerId = applicationId; final List<ScopedRowKey<CollectionPrefixedKey<Id>>> rowKeys = new ArrayList<>( entityIds.size() ); for ( final Id entityId : entityIds ) { final String collectionName = LegacyScopeUtils.getCollectionScopeNameFromEntityType( entityId.getType() ); final CollectionPrefixedKey<Id> collectionPrefixedKey = new CollectionPrefixedKey<>( collectionName, ownerId, entityId ); final ScopedRowKey<CollectionPrefixedKey<Id>> rowKey = ScopedRowKey.fromKey( applicationId, collectionPrefixedKey ); rowKeys.add( rowKey ); } final int maxEntityResultSizeInBytes = serializationFig.getMaxEntitySize() * entityIds.size(); final int numberRequests = Math.max(1, maxEntityResultSizeInBytes / cassandraFig.getThriftBufferSize()); final int entitiesPerRequest = entityIds.size() / numberRequests; final Scheduler scheduler; if(numberRequests == 1){ scheduler = Schedulers.immediate(); } else{ scheduler = Schedulers.io(); } final EntitySetImpl entitySetResults = Observable.from( rowKeys ) .buffer( entitiesPerRequest ).flatMap( listObservable -> { return Observable.just( listObservable ).map( scopedRowKeys -> { try { return keyspace.prepareQuery( columnFamily ).getKeySlice( scopedRowKeys ) .withColumnRange( maxVersion, null, false, 1 ).execute().getResult(); } catch ( ConnectionException e ) { throw new CollectionRuntimeException( null, applicationScope, "An error occurred connecting to cassandra", e ); } } ).subscribeOn( scheduler ); }, 10 ) .reduce( new EntitySetImpl( entityIds.size() ), ( entitySet, rows ) -> { final Iterator<Row<ScopedRowKey<CollectionPrefixedKey<Id>>, UUID>> latestEntityColumns = rows.iterator(); while ( latestEntityColumns.hasNext() ) { final Row<ScopedRowKey<CollectionPrefixedKey<Id>>, UUID> row = latestEntityColumns.next(); final ColumnList<UUID> columns = row.getColumns(); if ( columns.size() == 0 ) { continue; } final Id entityId = row.getKey().getKey().getSubKey(); final Column<UUID> column = columns.getColumnByIndex( 0 ); final MvccEntity parsedEntity = new MvccColumnParser( entityId, getEntitySerializer() ).parseColumn( column ); entitySet.addEntity( parsedEntity ); } return entitySet; } ).toBlocking().last(); return entitySetResults; }
MvccEntitySerializationStrategyImpl implements MvccEntitySerializationStrategy { @Override public EntitySet load( final ApplicationScope applicationScope, final Collection<Id> entityIds, final UUID maxVersion ) { Preconditions.checkNotNull( applicationScope, "applicationScope is required" ); Preconditions.checkNotNull( entityIds, "entityIds is required" ); Preconditions.checkArgument( entityIds.size() > 0, "entityIds is required" ); Preconditions.checkNotNull( maxVersion, "version is required" ); if( entityIds.size() > serializationFig.getMaxLoadSize()){ throw new IllegalArgumentException( "requested load size cannot be over configured maximum of " + serializationFig.getMaxLoadSize() ); } final Id applicationId = applicationScope.getApplication(); final Id ownerId = applicationId; final List<ScopedRowKey<CollectionPrefixedKey<Id>>> rowKeys = new ArrayList<>( entityIds.size() ); for ( final Id entityId : entityIds ) { final String collectionName = LegacyScopeUtils.getCollectionScopeNameFromEntityType( entityId.getType() ); final CollectionPrefixedKey<Id> collectionPrefixedKey = new CollectionPrefixedKey<>( collectionName, ownerId, entityId ); final ScopedRowKey<CollectionPrefixedKey<Id>> rowKey = ScopedRowKey.fromKey( applicationId, collectionPrefixedKey ); rowKeys.add( rowKey ); } final int maxEntityResultSizeInBytes = serializationFig.getMaxEntitySize() * entityIds.size(); final int numberRequests = Math.max(1, maxEntityResultSizeInBytes / cassandraFig.getThriftBufferSize()); final int entitiesPerRequest = entityIds.size() / numberRequests; final Scheduler scheduler; if(numberRequests == 1){ scheduler = Schedulers.immediate(); } else{ scheduler = Schedulers.io(); } final EntitySetImpl entitySetResults = Observable.from( rowKeys ) .buffer( entitiesPerRequest ).flatMap( listObservable -> { return Observable.just( listObservable ).map( scopedRowKeys -> { try { return keyspace.prepareQuery( columnFamily ).getKeySlice( scopedRowKeys ) .withColumnRange( maxVersion, null, false, 1 ).execute().getResult(); } catch ( ConnectionException e ) { throw new CollectionRuntimeException( null, applicationScope, "An error occurred connecting to cassandra", e ); } } ).subscribeOn( scheduler ); }, 10 ) .reduce( new EntitySetImpl( entityIds.size() ), ( entitySet, rows ) -> { final Iterator<Row<ScopedRowKey<CollectionPrefixedKey<Id>>, UUID>> latestEntityColumns = rows.iterator(); while ( latestEntityColumns.hasNext() ) { final Row<ScopedRowKey<CollectionPrefixedKey<Id>>, UUID> row = latestEntityColumns.next(); final ColumnList<UUID> columns = row.getColumns(); if ( columns.size() == 0 ) { continue; } final Id entityId = row.getKey().getKey().getSubKey(); final Column<UUID> column = columns.getColumnByIndex( 0 ); final MvccEntity parsedEntity = new MvccColumnParser( entityId, getEntitySerializer() ).parseColumn( column ); entitySet.addEntity( parsedEntity ); } return entitySet; } ).toBlocking().last(); return entitySetResults; } }
MvccEntitySerializationStrategyImpl implements MvccEntitySerializationStrategy { @Override public EntitySet load( final ApplicationScope applicationScope, final Collection<Id> entityIds, final UUID maxVersion ) { Preconditions.checkNotNull( applicationScope, "applicationScope is required" ); Preconditions.checkNotNull( entityIds, "entityIds is required" ); Preconditions.checkArgument( entityIds.size() > 0, "entityIds is required" ); Preconditions.checkNotNull( maxVersion, "version is required" ); if( entityIds.size() > serializationFig.getMaxLoadSize()){ throw new IllegalArgumentException( "requested load size cannot be over configured maximum of " + serializationFig.getMaxLoadSize() ); } final Id applicationId = applicationScope.getApplication(); final Id ownerId = applicationId; final List<ScopedRowKey<CollectionPrefixedKey<Id>>> rowKeys = new ArrayList<>( entityIds.size() ); for ( final Id entityId : entityIds ) { final String collectionName = LegacyScopeUtils.getCollectionScopeNameFromEntityType( entityId.getType() ); final CollectionPrefixedKey<Id> collectionPrefixedKey = new CollectionPrefixedKey<>( collectionName, ownerId, entityId ); final ScopedRowKey<CollectionPrefixedKey<Id>> rowKey = ScopedRowKey.fromKey( applicationId, collectionPrefixedKey ); rowKeys.add( rowKey ); } final int maxEntityResultSizeInBytes = serializationFig.getMaxEntitySize() * entityIds.size(); final int numberRequests = Math.max(1, maxEntityResultSizeInBytes / cassandraFig.getThriftBufferSize()); final int entitiesPerRequest = entityIds.size() / numberRequests; final Scheduler scheduler; if(numberRequests == 1){ scheduler = Schedulers.immediate(); } else{ scheduler = Schedulers.io(); } final EntitySetImpl entitySetResults = Observable.from( rowKeys ) .buffer( entitiesPerRequest ).flatMap( listObservable -> { return Observable.just( listObservable ).map( scopedRowKeys -> { try { return keyspace.prepareQuery( columnFamily ).getKeySlice( scopedRowKeys ) .withColumnRange( maxVersion, null, false, 1 ).execute().getResult(); } catch ( ConnectionException e ) { throw new CollectionRuntimeException( null, applicationScope, "An error occurred connecting to cassandra", e ); } } ).subscribeOn( scheduler ); }, 10 ) .reduce( new EntitySetImpl( entityIds.size() ), ( entitySet, rows ) -> { final Iterator<Row<ScopedRowKey<CollectionPrefixedKey<Id>>, UUID>> latestEntityColumns = rows.iterator(); while ( latestEntityColumns.hasNext() ) { final Row<ScopedRowKey<CollectionPrefixedKey<Id>>, UUID> row = latestEntityColumns.next(); final ColumnList<UUID> columns = row.getColumns(); if ( columns.size() == 0 ) { continue; } final Id entityId = row.getKey().getKey().getSubKey(); final Column<UUID> column = columns.getColumnByIndex( 0 ); final MvccEntity parsedEntity = new MvccColumnParser( entityId, getEntitySerializer() ).parseColumn( column ); entitySet.addEntity( parsedEntity ); } return entitySet; } ).toBlocking().last(); return entitySetResults; } @Inject  MvccEntitySerializationStrategyImpl( final Keyspace keyspace, final SerializationFig serializationFig,
                                                final CassandraFig cassandraFig ); }
MvccEntitySerializationStrategyImpl implements MvccEntitySerializationStrategy { @Override public EntitySet load( final ApplicationScope applicationScope, final Collection<Id> entityIds, final UUID maxVersion ) { Preconditions.checkNotNull( applicationScope, "applicationScope is required" ); Preconditions.checkNotNull( entityIds, "entityIds is required" ); Preconditions.checkArgument( entityIds.size() > 0, "entityIds is required" ); Preconditions.checkNotNull( maxVersion, "version is required" ); if( entityIds.size() > serializationFig.getMaxLoadSize()){ throw new IllegalArgumentException( "requested load size cannot be over configured maximum of " + serializationFig.getMaxLoadSize() ); } final Id applicationId = applicationScope.getApplication(); final Id ownerId = applicationId; final List<ScopedRowKey<CollectionPrefixedKey<Id>>> rowKeys = new ArrayList<>( entityIds.size() ); for ( final Id entityId : entityIds ) { final String collectionName = LegacyScopeUtils.getCollectionScopeNameFromEntityType( entityId.getType() ); final CollectionPrefixedKey<Id> collectionPrefixedKey = new CollectionPrefixedKey<>( collectionName, ownerId, entityId ); final ScopedRowKey<CollectionPrefixedKey<Id>> rowKey = ScopedRowKey.fromKey( applicationId, collectionPrefixedKey ); rowKeys.add( rowKey ); } final int maxEntityResultSizeInBytes = serializationFig.getMaxEntitySize() * entityIds.size(); final int numberRequests = Math.max(1, maxEntityResultSizeInBytes / cassandraFig.getThriftBufferSize()); final int entitiesPerRequest = entityIds.size() / numberRequests; final Scheduler scheduler; if(numberRequests == 1){ scheduler = Schedulers.immediate(); } else{ scheduler = Schedulers.io(); } final EntitySetImpl entitySetResults = Observable.from( rowKeys ) .buffer( entitiesPerRequest ).flatMap( listObservable -> { return Observable.just( listObservable ).map( scopedRowKeys -> { try { return keyspace.prepareQuery( columnFamily ).getKeySlice( scopedRowKeys ) .withColumnRange( maxVersion, null, false, 1 ).execute().getResult(); } catch ( ConnectionException e ) { throw new CollectionRuntimeException( null, applicationScope, "An error occurred connecting to cassandra", e ); } } ).subscribeOn( scheduler ); }, 10 ) .reduce( new EntitySetImpl( entityIds.size() ), ( entitySet, rows ) -> { final Iterator<Row<ScopedRowKey<CollectionPrefixedKey<Id>>, UUID>> latestEntityColumns = rows.iterator(); while ( latestEntityColumns.hasNext() ) { final Row<ScopedRowKey<CollectionPrefixedKey<Id>>, UUID> row = latestEntityColumns.next(); final ColumnList<UUID> columns = row.getColumns(); if ( columns.size() == 0 ) { continue; } final Id entityId = row.getKey().getKey().getSubKey(); final Column<UUID> column = columns.getColumnByIndex( 0 ); final MvccEntity parsedEntity = new MvccColumnParser( entityId, getEntitySerializer() ).parseColumn( column ); entitySet.addEntity( parsedEntity ); } return entitySet; } ).toBlocking().last(); return entitySetResults; } @Inject  MvccEntitySerializationStrategyImpl( final Keyspace keyspace, final SerializationFig serializationFig,
                                                final CassandraFig cassandraFig ); @Override MutationBatch write( final ApplicationScope applicationScope, final MvccEntity entity ); @Override EntitySet load( final ApplicationScope applicationScope, final Collection<Id> entityIds,
                           final UUID maxVersion ); @Override Iterator<MvccEntity> loadDescendingHistory( final ApplicationScope applicationScope, final Id entityId,
                                                       final UUID version, final int fetchSize ); @Override Iterator<MvccEntity> loadAscendingHistory( final ApplicationScope applicationScope, final Id entityId,
                                                      final UUID version, final int fetchSize ); @Override Optional<MvccEntity> load( final ApplicationScope scope, final Id entityId ); @Override MutationBatch mark( final ApplicationScope applicationScope, final Id entityId, final UUID version ); @Override MutationBatch delete( final ApplicationScope applicationScope, final Id entityId, final UUID version ); @Override java.util.Collection getColumnFamilies(); @Override Collection<TableDefinition> getTables(); }
MvccEntitySerializationStrategyImpl implements MvccEntitySerializationStrategy { @Override public EntitySet load( final ApplicationScope applicationScope, final Collection<Id> entityIds, final UUID maxVersion ) { Preconditions.checkNotNull( applicationScope, "applicationScope is required" ); Preconditions.checkNotNull( entityIds, "entityIds is required" ); Preconditions.checkArgument( entityIds.size() > 0, "entityIds is required" ); Preconditions.checkNotNull( maxVersion, "version is required" ); if( entityIds.size() > serializationFig.getMaxLoadSize()){ throw new IllegalArgumentException( "requested load size cannot be over configured maximum of " + serializationFig.getMaxLoadSize() ); } final Id applicationId = applicationScope.getApplication(); final Id ownerId = applicationId; final List<ScopedRowKey<CollectionPrefixedKey<Id>>> rowKeys = new ArrayList<>( entityIds.size() ); for ( final Id entityId : entityIds ) { final String collectionName = LegacyScopeUtils.getCollectionScopeNameFromEntityType( entityId.getType() ); final CollectionPrefixedKey<Id> collectionPrefixedKey = new CollectionPrefixedKey<>( collectionName, ownerId, entityId ); final ScopedRowKey<CollectionPrefixedKey<Id>> rowKey = ScopedRowKey.fromKey( applicationId, collectionPrefixedKey ); rowKeys.add( rowKey ); } final int maxEntityResultSizeInBytes = serializationFig.getMaxEntitySize() * entityIds.size(); final int numberRequests = Math.max(1, maxEntityResultSizeInBytes / cassandraFig.getThriftBufferSize()); final int entitiesPerRequest = entityIds.size() / numberRequests; final Scheduler scheduler; if(numberRequests == 1){ scheduler = Schedulers.immediate(); } else{ scheduler = Schedulers.io(); } final EntitySetImpl entitySetResults = Observable.from( rowKeys ) .buffer( entitiesPerRequest ).flatMap( listObservable -> { return Observable.just( listObservable ).map( scopedRowKeys -> { try { return keyspace.prepareQuery( columnFamily ).getKeySlice( scopedRowKeys ) .withColumnRange( maxVersion, null, false, 1 ).execute().getResult(); } catch ( ConnectionException e ) { throw new CollectionRuntimeException( null, applicationScope, "An error occurred connecting to cassandra", e ); } } ).subscribeOn( scheduler ); }, 10 ) .reduce( new EntitySetImpl( entityIds.size() ), ( entitySet, rows ) -> { final Iterator<Row<ScopedRowKey<CollectionPrefixedKey<Id>>, UUID>> latestEntityColumns = rows.iterator(); while ( latestEntityColumns.hasNext() ) { final Row<ScopedRowKey<CollectionPrefixedKey<Id>>, UUID> row = latestEntityColumns.next(); final ColumnList<UUID> columns = row.getColumns(); if ( columns.size() == 0 ) { continue; } final Id entityId = row.getKey().getKey().getSubKey(); final Column<UUID> column = columns.getColumnByIndex( 0 ); final MvccEntity parsedEntity = new MvccColumnParser( entityId, getEntitySerializer() ).parseColumn( column ); entitySet.addEntity( parsedEntity ); } return entitySet; } ).toBlocking().last(); return entitySetResults; } @Inject  MvccEntitySerializationStrategyImpl( final Keyspace keyspace, final SerializationFig serializationFig,
                                                final CassandraFig cassandraFig ); @Override MutationBatch write( final ApplicationScope applicationScope, final MvccEntity entity ); @Override EntitySet load( final ApplicationScope applicationScope, final Collection<Id> entityIds,
                           final UUID maxVersion ); @Override Iterator<MvccEntity> loadDescendingHistory( final ApplicationScope applicationScope, final Id entityId,
                                                       final UUID version, final int fetchSize ); @Override Iterator<MvccEntity> loadAscendingHistory( final ApplicationScope applicationScope, final Id entityId,
                                                      final UUID version, final int fetchSize ); @Override Optional<MvccEntity> load( final ApplicationScope scope, final Id entityId ); @Override MutationBatch mark( final ApplicationScope applicationScope, final Id entityId, final UUID version ); @Override MutationBatch delete( final ApplicationScope applicationScope, final Id entityId, final UUID version ); @Override java.util.Collection getColumnFamilies(); @Override Collection<TableDefinition> getTables();  }