我们如何将lagom的Read-side processor与Dgraph一起使用?
How can we use lagom's Read-side processor with Dgraph?
我是lagom和dgraph的新手。我被困在如何将 lagom 的读取端处理器与 Dgraph 一起使用。下面是将 Cassandra 与 lagom 结合使用的代码。
import akka.NotUsed;
import com.lightbend.lagom.javadsl.api.ServiceCall;
import com.lightbend.lagom.javadsl.persistence.cassandra.CassandraSession;
import java.util.concurrent.CompletableFuture;
import javax.inject.Inject;
import akka.stream.javadsl.Source;
public class FriendServiceImpl implements FriendService {
private final CassandraSession cassandraSession;
@Inject
public FriendServiceImpl(CassandraSession cassandraSession) {
this.cassandraSession = cassandraSession;
}
//Implement your service method here
}
Lagom 不提供对 Dgraph 的开箱即用支持。如果你必须使用Lagom的Read-Side processor with Dgraph,那么你必须使用Lagom的Generic Read Side support。像这样:
/**
* Read side processor for Dgraph.
*/
public class FriendEventProcessor extends ReadSideProcessor<FriendEvent> {
private static void createModel() {
//TODO: Initialize schema in Dgraph
}
@Override
public ReadSideProcessor.ReadSideHandler<FriendEvent> buildHandler() {
return new ReadSideHandler<FriendEvent>() {
private final Done doneInstance = Done.getInstance();
@Override
public CompletionStage<Done> globalPrepare() {
createModel();
return CompletableFuture.completedFuture(doneInstance);
}
@Override
public CompletionStage<Offset> prepare(final AggregateEventTag<FriendEvent> tag) {
return CompletableFuture.completedFuture(Offset.NONE);
}
@Override
public Flow<Pair<FriendEvent, Offset>, Done, ?> handle() {
return Flow.<Pair<FriendEvent, Offset>>create()
.mapAsync(1, eventAndOffset -> {
if (eventAndOffset.first() instanceof FriendCreated) {
//TODO: Add Friend in Dgraph;
}
return CompletableFuture.completedFuture(doneInstance);
}
);
}
};
}
@Override
public PSequence<AggregateEventTag<FriendEvent>> aggregateTags() {
return FriendEvent.TAG.allTags();
}
}
对于FriendEvent.TAG.allTags()
,您必须在FriendEvent
接口中添加以下代码:
int NUM_SHARDS = 20;
AggregateEventShards<FriendEvent> TAG =
AggregateEventTag.sharded(FriendEvent.class, NUM_SHARDS);
@Override
default AggregateEventShards<FriendEvent> aggregateTag() {
return TAG;
}
希望对您有所帮助!
我是lagom和dgraph的新手。我被困在如何将 lagom 的读取端处理器与 Dgraph 一起使用。下面是将 Cassandra 与 lagom 结合使用的代码。
import akka.NotUsed;
import com.lightbend.lagom.javadsl.api.ServiceCall;
import com.lightbend.lagom.javadsl.persistence.cassandra.CassandraSession;
import java.util.concurrent.CompletableFuture;
import javax.inject.Inject;
import akka.stream.javadsl.Source;
public class FriendServiceImpl implements FriendService {
private final CassandraSession cassandraSession;
@Inject
public FriendServiceImpl(CassandraSession cassandraSession) {
this.cassandraSession = cassandraSession;
}
//Implement your service method here
}
Lagom 不提供对 Dgraph 的开箱即用支持。如果你必须使用Lagom的Read-Side processor with Dgraph,那么你必须使用Lagom的Generic Read Side support。像这样:
/**
* Read side processor for Dgraph.
*/
public class FriendEventProcessor extends ReadSideProcessor<FriendEvent> {
private static void createModel() {
//TODO: Initialize schema in Dgraph
}
@Override
public ReadSideProcessor.ReadSideHandler<FriendEvent> buildHandler() {
return new ReadSideHandler<FriendEvent>() {
private final Done doneInstance = Done.getInstance();
@Override
public CompletionStage<Done> globalPrepare() {
createModel();
return CompletableFuture.completedFuture(doneInstance);
}
@Override
public CompletionStage<Offset> prepare(final AggregateEventTag<FriendEvent> tag) {
return CompletableFuture.completedFuture(Offset.NONE);
}
@Override
public Flow<Pair<FriendEvent, Offset>, Done, ?> handle() {
return Flow.<Pair<FriendEvent, Offset>>create()
.mapAsync(1, eventAndOffset -> {
if (eventAndOffset.first() instanceof FriendCreated) {
//TODO: Add Friend in Dgraph;
}
return CompletableFuture.completedFuture(doneInstance);
}
);
}
};
}
@Override
public PSequence<AggregateEventTag<FriendEvent>> aggregateTags() {
return FriendEvent.TAG.allTags();
}
}
对于FriendEvent.TAG.allTags()
,您必须在FriendEvent
接口中添加以下代码:
int NUM_SHARDS = 20;
AggregateEventShards<FriendEvent> TAG =
AggregateEventTag.sharded(FriendEvent.class, NUM_SHARDS);
@Override
default AggregateEventShards<FriendEvent> aggregateTag() {
return TAG;
}
希望对您有所帮助!