No. of Instances - 11
No. of Commits - 6
No. of Projects - {'flink'}
Hierarchy/Composition: -
Primitive Info: -
NameSpace: Jdk -> Jdk
Mapping:
- {
@Override public void call( List<Edge> edges){
final MutationBatch batch=keyspace.prepareMutationBatch();
for ( Edge edge : edges) {
logger.info("Migrating meta for edge {}",edge);
final MutationBatch edgeBatch=edgeMigrationStrategy.getMigration().to().writeEdge(applicationScope,edge);
batch.mergeShallow(edgeBatch);
}
try {
batch.execute();
}
catch ( ConnectionException e) {
throw new RuntimeException("Unable to perform migration",e);
}
final long newCount=counter.addAndGet(edges.size());
observer.update(getVersion(),String.format("Currently running. Rewritten %d edge types",newCount));
}
}
to {
@Override public void call( List<Edge> edges){
final MutationBatch batch=keyspace.prepareMutationBatch();
for ( Edge edge : edges) {
logger.info("Migrating meta for edge {}",edge);
final MutationBatch edgeBatch=migration.to.writeEdge(applicationScope,edge);
batch.mergeShallow(edgeBatch);
}
try {
batch.execute();
}
catch ( ConnectionException e) {
throw new RuntimeException("Unable to perform migration",e);
}
final long newCount=counter.addAndGet(edges.size());
observer.update(migration.to.getImplementationVersion(),String.format("Currently running. Rewritten %d edge types",newCount));
}
}
- {
@Override public Observable call( final Observable<List<Edge>> listObservable){
return listObservable.doOnNext(new Action1<List<Edge>>(){
@Override public void call( List<Edge> edges){
final MutationBatch batch=keyspace.prepareMutationBatch();
for ( Edge edge : edges) {
logger.info("Migrating meta for edge {}",edge);
final MutationBatch edgeBatch=edgeMigrationStrategy.getMigration().to().writeEdge(applicationScope,edge);
batch.mergeShallow(edgeBatch);
}
try {
batch.execute();
}
catch ( ConnectionException e) {
throw new RuntimeException("Unable to perform migration",e);
}
final long newCount=counter.addAndGet(edges.size());
observer.update(getVersion(),String.format("Currently running. Rewritten %d edge types",newCount));
}
}
);
}
}
to {
@Override public Observable call( final Observable<List<Edge>> listObservable){
return listObservable.doOnNext(new Action1<List<Edge>>(){
@Override public void call( List<Edge> edges){
final MutationBatch batch=keyspace.prepareMutationBatch();
for ( Edge edge : edges) {
logger.info("Migrating meta for edge {}",edge);
final MutationBatch edgeBatch=migration.to.writeEdge(applicationScope,edge);
batch.mergeShallow(edgeBatch);
}
try {
batch.execute();
}
catch ( ConnectionException e) {
throw new RuntimeException("Unable to perform migration",e);
}
final long newCount=counter.addAndGet(edges.size());
observer.update(migration.to.getImplementationVersion(),String.format("Currently running. Rewritten %d edge types",newCount));
}
}
);
}
}