Skip to content

add field argument to newDRPCStream #1258

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
lujinhong wants to merge 1 commit into from
Closed

add field argument to newDRPCStream #1258

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 15 additions & 5 deletions storm-core/src/jvm/org/apache/storm/trident/TridentTopology.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,8 @@ public class TridentTopology {
final DefaultDirectedGraph<Node, IndexedEdge> _graph;
final Map<String, List<Node>> _colocate;
final UniqueIdGen _gen;
private final static String DRPC_DEFAULT_FILED = "args";


public TridentTopology() {
this(new DefaultDirectedGraph<Node, IndexedEdge>(new ErrorEdgeFactory()),
Expand Down Expand Up @@ -145,27 +147,35 @@ public Stream newStream(String txId, ITridentDataSource dataSource) {
}

public Stream newDRPCStream(String function) {
return newDRPCStream(new DRPCSpout(function));
return newDRPCStream (new DRPCSpout(function), DRPC_DEFAULT_FILED);
}


public Stream newDRPCStream(String function, String field) {
return newDRPCStream (new DRPCSpout(function), field);
}

public Stream newDRPCStream(String function, ILocalDRPC server) {
return newDRPCStream (function, server, DRPC_DEFAULT_FILED);
}

public Stream newDRPCStream(String function, ILocalDRPC server, String field) {
DRPCSpout spout;
if(server==null) {
spout = new DRPCSpout(function);
} else {
spout = new DRPCSpout(function, server);
}
return newDRPCStream(spout);
return newDRPCStream (spout, field);
}

private Stream newDRPCStream(DRPCSpout spout) {
private Stream newDRPCStream(DRPCSpout spout, String field) {
// TODO: consider adding a shuffle grouping after the spout to avoid so much routing of the args/return-info all over the place
// (at least until its possible to just pack bolt logic into the spout itself)

Node n = new SpoutNode(getUniqueStreamId(), TridentUtils.getSingleOutputStreamFields(spout), null, spout, SpoutNode.SpoutType.DRPC);
Stream nextStream = addNode(n);
// later on, this will be joined back with return-info and all the results
return nextStream.project(new Fields("args"));
return nextStream.project(new Fields(field));
}

public TridentState newStaticState(StateFactory factory) {
Expand Down