Java 类org.hibernate.ogm.dialect.spi.ModelConsumer 实例源码

@Override
public void forEachTuple(final ModelConsumer consumer, TupleTypeContext tupleTypeContext, EntityKeyMetadata entityKeyMetadata) {
    KeyScanCursor<String> cursor = null;
    String prefix = entityKeyMetadata.getTable() + ":";

    ScanArgs scanArgs = ScanArgs.Builder.matches( prefix + "*" );
    do {
        cursor = scan( cursor, scanArgs );
        consumer.consume( new RedisJsonDialectTuplesSupplier( cursor, entityStorageStrategy, prefix, entityKeyMetadata ) );
    } while ( !cursor.isFinished() );
}

@Override
public void forEachTuple(ModelConsumer consumer, TupleTypeContext tupleTypeContext, EntityKeyMetadata entityKeyMetadata) {
    KeyScanCursor<String> cursor = null;
    String prefix = entityKeyMetadata.getTable() + ":";

    ScanArgs scanArgs = ScanArgs.Builder.matches( prefix + "*" );
    do {
        cursor = scan( cursor, scanArgs );
        consumer.consume( new RedisHashDialectTuplesSupplier( cursor, connection, prefix, entityKeyMetadata ) );
    } while ( !cursor.isFinished() );
}

@Test
public void testScan() throws Exception {

    AbstractRedisDialect dialect = RedisTestHelper.getDialect( getProvider() );
    assumeTrue( dialect.isClusterMode() );

    // pre-computed key file.
    URL resource = Resources.getResource( "redis-cluster-slothashes.txt" );
    List<String> lines = Resources.readLines( resource, StandardCharsets.ISO_8859_1 );

    OgmSession session = openSession();
    session.getTransaction().begin();

    // given
    int availableKeys = 0;
    for ( String line : lines ) {

        if ( line.startsWith( "#" ) || line.trim().isEmpty() ) {
            continue;
        }

        String key = line.substring( 0, line.indexOf( ' ' ) ).trim();

        Band record = new Band( key, key );
        session.persist( record );
        availableKeys++;
    }
    session.getTransaction().commit();

    final AtomicInteger counter = new AtomicInteger();

    dialect.forEachTuple( new ModelConsumer() {
        @Override
        public void consume(TuplesSupplier supplier) {
            try ( ClosableIterator<Tuple> closableIterator = supplier.get( null ) ) {
                while ( closableIterator.hasNext() ) {
                    counter.incrementAndGet();
                }
            }
        }
    }, null, new DefaultEntityKeyMetadata( "Band", new String[] {"id"} ) );

    assertEquals( availableKeys, counter.get() );
}

@Override
public void forEachTuple(ModelConsumer consumer, TupleTypeContext tupleTypeContext, EntityKeyMetadata entityKeyMetadata) {
    throw new UnsupportedOperationException( "forEachTuple() is not implemented" );
}

@Override
public void forEachTuple(ModelConsumer consumer, EntityKeyMetadata... entityKeyMetadatas) {
    throw new UnsupportedOperationException("not yet supported");
}