Source code

001package com.hammurapi.common;
002
003import java.lang.reflect.Array;
004import java.util.ArrayList;
005import java.util.Collection;
006import java.util.HashMap;
007import java.util.Iterator;
008import java.util.List;
009import java.util.Map;
010import java.util.concurrent.ConcurrentHashMap;
011import java.util.logging.Level;
012import java.util.logging.Logger;
013
014import com.hammurapi.extract.CommutativeAnd;
015import com.hammurapi.extract.Extractor;
016import com.hammurapi.extract.Predicate;
017
018/**
019 * Helper class to join data from several inputs.
020 * @author Pavel Vlasov
021 * @param <T> Join element type.
022 * @param <C> Context type passed from addInput to join.
023 * @param <R> Return type passed from join to to addInput.
024 */
025public abstract class Joiner<T, C, R> {   
026        
027        /**
028         * Collector of objects to join. Implements a subset of collection 
029         * methods. 
030         * @author Pavel Vlasov
031         *
032         * @param <T>
033         */
034        public interface Collector<T> extends Iterable<T> {
035                                
036                /**
037                 * Adds new object to the collector. This method shall not break currently open iterators.
038                 * @param obj
039                 * @return true if object was added to the collector.
040                 */
041                boolean add(T obj);
042                
043                /**
044                 * Removes object (last entry) from the collector.
045                 * @param obj
046                 * @return true if success
047                 */
048                boolean remove(T obj);
049                
050                /**
051                 * Removes all object from the collector.
052                 */
053                void clear();
054                
055        }
056        
057        /**
058         * Adapter for collections.
059         * @author Pavel Vlasov
060         *
061         * @param <T>
062         */
063        public static class CollectionAdapter<T> implements Collector<T> {
064                
065                private Collection<T> master;
066                
067                public CollectionAdapter(Collection<T> master) {
068                        this.master = master;
069                }
070
071                public boolean add(T obj) {
072                        return master.add(obj);
073                }
074
075                public void clear() {
076                        master.clear();                 
077                }
078
079                public boolean remove(T obj) {
080                        return master.remove(obj);                      
081                }
082
083                public Iterator<T> iterator() {
084                        return master.iterator();
085                }
086
087                @Override
088                public String toString() {
089                        return "CollectionAdapter [master=" + master + "]";
090                }
091                
092        }
093        
094        private static final Logger logger = Logger.getLogger(Joiner.class.getName());
095        
096        /**
097         * Interface to consume inputs in join() method.
098         * @author Pavel Vlasov
099         */
100        public static class InputConsumer {
101                
102                boolean[] consumeFlags;
103
104                InputConsumer(int size) {
105                        consumeFlags = new boolean[size];
106                }
107                
108                /**
109                 * Consumes input at given index.
110                 * @param index
111                 */
112                public void consume(int index) {
113                        consumeFlags[index] = true;
114                }
115        }
116        
117        private Collector<T>[] inputCollectors;
118        private boolean outerJoin;
119        private Class<T> inputType;
120        private boolean isFine;
121        
122        /**
123         * Predicates to be fired at a particular index.
124         */
125        private Map<Integer, List<Predicate<T, C>>> predicates = new HashMap<Integer, List<Predicate<T, C>>>();
126        
127        /**
128         * Adds join predicate.
129         * @param predicate
130         */
131        public void addPredicate(Predicate<T, C> predicate) {
132                if (predicate instanceof CommutativeAnd) {
133                        // Break down the predicate.
134                        for (Predicate<T,C> part: ((CommutativeAnd<T,C>) predicate).getParts()) {
135                                addPredicate(part);
136                        }               
137                } else {                
138                        int maxIndex = 0;
139                        for (Integer idx: predicate.parameterIndices()) {
140                                if (idx>maxIndex) {
141                                        maxIndex = idx;
142                                }
143                        }
144                        List<Predicate<T, C>> pl = predicates.get(maxIndex);
145                        if (pl==null) {
146                                pl = new ArrayList<Predicate<T, C>>();
147                                predicates.put(maxIndex, pl);
148                        }
149                        pl.add(predicate);
150                }
151        }
152
153        /**
154         * Creates joiner.
155         * @param inputCollectors Input collectors.
156         * @param inputType Input type is required because of erasure of generics and T is needed at runtime.
157         * @param outerJoin If true, each input addition triggers invocation of join, even if other input collectors
158         * don't have any data.
159         */
160        protected Joiner(Collector<T>[] inputCollectors, Class<T> inputType, boolean outerJoin) {
161                this.inputCollectors = inputCollectors;
162                this.outerJoin = outerJoin;
163                this.inputType = inputType;
164                this.isFine = logger.isLoggable(Level.FINE);
165        }
166        
167        /**
168         * Start join is invoked before join. Typically this method shall
169         * obtain a lock on input collectors.
170         */
171        protected abstract void startJoin();
172        
173        /**
174         * This method is invoked after join is finished. Typically this method
175         * shall release the lock on input collectors.
176         */
177        protected abstract void endJoin();
178        
179        /**
180         * Adds input. Input is joined with other accumulated inputs.
181         *  
182         * @param index Input index. 
183         * @param input Input.
184         */
185        @SuppressWarnings("unchecked")
186        public List<R> addInput(int index, T input, C context) throws Exception {
187                if (isFine) {
188                        logger.fine("New join input: "+input+" at index "+index);
189                        StringBuilder sb = new StringBuilder();
190                        sb.append("Input collectors: ");
191                        for (Collector<T> c:inputCollectors) {
192                                sb.append(c+" ");
193                        }
194                        logger.fine(sb.toString());
195                }
196                if (isValidInput(index, input)) {
197                        startJoin();
198                        try {
199                                if (inputCollectors[index].add(input)) { // Join only if successfully added - for sets.
200                                        T[] inputs = (T[]) Array.newInstance(inputType, inputCollectors.length);
201                                        inputs[index] = input;
202                                        InputConsumer consumer = new InputConsumer(inputCollectors.length);                     
203                                        List<R> ret = new ArrayList<R>();
204                                        Map<C, Map<Extractor<T, ? super Boolean, C>, ? super Boolean>> cache = new ConcurrentHashMap<C, Map<Extractor<T,? super Boolean,C>,? super Boolean>>(); 
205                                        join(inputs, 0, index, context, consumer, cache , ret);
206                                        if (consumer.consumeFlags[index]) { // Current input was consumed.
207                                                if (!inputCollectors[index].remove(input)) {
208                                                        throw new IllegalStateException("Cannot remove just added input");
209                                                }
210                                        }
211                                        return ret;
212                                }
213                        } finally {
214                                endJoin();
215                        }
216                }
217                return null;
218        }
219        
220        /**
221         * This method can be overridden to implement concurrent joining, i.e. at particular indices the sub-method wraps call to the super-method
222         * into a task executed in a separate thread.
223         * @param inputs
224         * @param currentIndex
225         * @param inputIndex
226         * @param context
227         * @param consumer
228         * @param cache
229         * @param resultCollector
230         * @throws Exception
231         */
232        protected void join(T[] inputs, int currentIndex, int inputIndex, C context, InputConsumer consumer, Map<C, Map<Extractor<T, ? super Boolean, C>, ? super Boolean>> cache, List<R> resultCollector) throws Exception {
233//              System.out.println(this+" Join "+currentIndex+" "+Arrays.toString(inputs));
234                if (currentIndex<inputs.length) {
235                        
236                        List<Predicate<T, C>> pl = predicates.get(currentIndex);
237                        if (pl!=null) {
238                                for (Predicate<T, C> predicate: pl) {
239                                        if (!predicate.extract(context, cache, inputs)) {
240                                                return;
241                                        }
242                                }
243                        }
244                        
245                        if (!partialJoin(inputs, context, consumer, currentIndex, inputIndex)) {
246                                return;
247                        }
248                }
249                
250                if (currentIndex==inputs.length) {
251                        for (int i=0; i<consumer.consumeFlags.length; ++i) {
252                                consumer.consumeFlags[i] = false;
253                        }
254                        R joinResult = join(inputs, context, consumer, inputIndex);
255                        synchronized (resultCollector) { // For concurrent joins.
256                                resultCollector.add(joinResult);
257                        }
258                } else if (currentIndex==inputIndex) {
259                        join(inputs, currentIndex+1, inputIndex, context, consumer, cache, resultCollector);
260                } else {                        
261                        if (outerJoin) {
262                                inputs[currentIndex]=null;
263                                join(inputs, currentIndex+1, inputIndex, context, consumer, cache, resultCollector);
264                                if (consumer.consumeFlags[inputIndex]) { // Input was consumed
265                                        return;
266                                }
267                                for (int i=0; i<currentIndex; ++i) { // Lower index input was consumed - get out of here.
268                                        if (consumer.consumeFlags[i]) {
269                                                return;
270                                        }
271                                }
272                        }
273                        
274                        Iterator<T> it = inputCollectors[currentIndex].iterator();
275                        try {
276                                while (it.hasNext()) {
277                                        inputs[currentIndex]=it.next();
278                                        if (isValidInput(currentIndex, inputs[currentIndex])) {
279                                                join(inputs, currentIndex+1, inputIndex, context, consumer, cache, resultCollector);
280                                                if (consumer.consumeFlags[currentIndex]) { // This input was consumed.
281                                                        it.remove();                                    
282                                                        return;
283                                                }
284                                                
285                                                if (consumer.consumeFlags[inputIndex]) { // Input was consumed
286                                                        return;
287                                                }
288                                                
289                                                for (int i=0; i<currentIndex; ++i) { // Lower index input was consumed - get out of here.
290                                                        if (consumer.consumeFlags[i]) {
291                                                                return;
292                                                        }
293                                                }
294                                        } else {
295                                                if (consumer.consumeFlags[currentIndex]) { // This input is not valid anymore.
296                                                        it.remove();                                    
297                                                        return;
298                                                }
299                                        }
300                                }
301                        } finally {
302                                if (it instanceof AutoCloseable) {
303                                        ((AutoCloseable) it).close();
304                                }
305                        }
306                }               
307        }
308        
309        /**
310         * This method is invoked for every combination of valid inputs.
311         * @param inputs Inputs
312         * @param context Join context passed from addInput
313         * @param consumer callback interface to consume inputs.
314         * @param activator Index of join activator.
315         * @return Join result. 
316         * @throws Exception
317         */
318        protected abstract R join(T[] inputs, C context, InputConsumer consumer, int activator) throws Exception;
319        
320        /**
321         * This method is invoked for all inputs before joining. It is useful for situation when 
322         * inputs in internal collections may become invalid over the course of life of joiner.
323         * @param index
324         * @param input
325         * @return
326         */
327        protected boolean isValidInput(int index, T input) {
328                return true;
329        }
330        
331        /**
332         * This method is invoked to validate already joined inputs.
333         * If this method returns false, further joining is abandoned.
334         * @param inputs Inputs array.
335         * @param index Index of last already joined input.
336         * @param consumer Callback interface to consume inputs. 
337         * @return
338         */
339        protected boolean partialJoin(T[] inputs, C context, InputConsumer consumer, int index, int activator) throws Exception {
340                return true;
341        }
342        
343        /**
344         * Clears join collections.
345         */
346        public void reset() {
347                for (Collector<T> collector: inputCollectors) {
348                        collector.clear();
349                }
350        }
351
352}