1
2 package org.galagosearch.tupleflow;
3
4 import java.io.IOException;
5 import java.lang.reflect.InvocationTargetException;
6 import java.util.Comparator;
7 import org.galagosearch.tupleflow.execution.ErrorHandler;
8 import org.galagosearch.tupleflow.execution.Verification;
9
10 /***
11 * Takes two inputs, keys and values. Each item in the values list that matches
12 * a key is sent to the next stage. Items are assumed to be sorted in order
13 * by "order". That order also determines equality.
14 *
15 * @author trevor
16 */
17 public class Select<T> implements ExNihiloSource<T> {
18 TypeReader<T> keys;
19 TypeReader<T> values;
20 public Processor<T> processor;
21 Class<T> klass;
22 Order<T> order;
23
24 public Select(TupleFlowParameters parameters) throws IOException, ClassNotFoundException,
25 NoSuchMethodException, IllegalArgumentException, InvocationTargetException,
26 InstantiationException, IllegalAccessException {
27 keys = parameters.getTypeReader("keys");
28 values = parameters.getTypeReader("values");
29 klass = (Class<T>) Class.forName(parameters.getXML().get("class"));
30 String orderSpec = parameters.getXML().get("order");
31
32 org.galagosearch.tupleflow.Type<T> type = (org.galagosearch.tupleflow.Type<T>) klass.
33 getConstructor().newInstance();
34 order = type.getOrder(orderSpec);
35 }
36
37 public void run() throws IOException {
38 T key = keys.read();
39 T value = values.read();
40 Comparator<T> lessThan = order.lessThan();
41
42 while (key != null && value != null) {
43 int compare = lessThan.compare(key, value);
44
45 if (compare < 0) {
46 key = keys.read();
47 } else if (compare > 0) {
48 value = values.read();
49 } else {
50 processor.process(value);
51 value = values.read();
52 }
53 }
54
55 while (key != null) {
56 key = keys.read();
57 }
58 while (value != null) {
59 value = values.read();
60 }
61 }
62
63 public Class<T> getOutputClass(TupleFlowParameters parameters) throws ClassNotFoundException {
64 return (Class<T>) Class.forName(parameters.getXML().get("class"));
65 }
66
67 public void setProcessor(Step next) throws IncompatibleProcessorException {
68 Linkage.link(this, next);
69 }
70
71 public static void verify(TupleFlowParameters parameters, ErrorHandler handler) throws ClassNotFoundException {
72 if (!Verification.requireParameters(new String[]{"class", "order"}, parameters.getXML(),
73 handler)) {
74 return;
75 }
76 String className = parameters.getXML().get("class");
77 String[] order = parameters.getXML().get("order").split(" ");
78 boolean result;
79
80 result = Verification.requireClass(className, handler);
81 result = result && Verification.requireOrder(className, order, handler);
82
83 if (!result) {
84 return;
85 }
86 Class klass = Class.forName(className);
87 Verification.verifyTypeReader("keys", klass, order, parameters, handler);
88 Verification.verifyTypeReader("values", klass, order, parameters, handler);
89 }
90 }