package com.hazelcast.impl;

import static com.hazelcast.impl.ConcurrentMapManager.Record;
import com.hazelcast.query.IndexAwarePredicate;
import com.hazelcast.query.IndexedPredicate;
import com.hazelcast.query.Predicate;
import com.hazelcast.query.RangedPredicate;
import static com.hazelcast.query.RangedPredicate.RangeType.*;

import java.util.*;
import java.util.logging.Logger;
import java.util.logging.Level;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class QueryService implements Runnable {

    private final Logger logger = Logger.getLogger(QueryService.class.getName());    
    final Node node;
    private volatile boolean running = true;

    final BlockingQueue<Runnable> queryQ = new LinkedBlockingQueue();

    public QueryService(Node node) {
        this.node = node;
    }

    public void run() {
        while (running) {
            Runnable run = null;
            try {
                run = queryQ.take();
                run.run();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }


    public void addNewIndex(final Index index, final long value, final Record record) {
        try {
            queryQ.put(new Runnable() {
                public void run() {
                    index.addNewIndex(value, record);
                }
            });
        } catch (InterruptedException e) {
        }
    }

    public void updateIndex(final Index index, final long oldValue, final long newValue, final Record record) {
        try {
            queryQ.put(new Runnable() {
                public void run() {
                    index.updateIndex(oldValue, newValue, record);
                }
            });
        } catch (InterruptedException e) {
        }
    }

    public void removeIndex(final Index index, final long value, final Record record) {
        try {
            queryQ.put(new Runnable() {
                public void run() {
                    index.removeIndex(value, record);
                }
            });
        } catch (InterruptedException e) {
        }
    }

    public static long getLongValue(Object value) {
        if (value instanceof Number) {
            return ((Number) value).longValue();
        } else if (value instanceof Boolean) {
            return (Boolean.TRUE.equals(value)) ? 1 : -1;
        } else {
            return value.hashCode();
        }
    }

    public Collection<Record> query(final Collection<Record> allRecords, final Map<String, Index> namedIndexes, final Predicate predicate) {
        try {
            final BlockingQueue<Collection<Record>> resultQ = new ArrayBlockingQueue(1);
            queryQ.put(new Runnable() {
                public void run() {
                    resultQ.offer(doQuery(allRecords, namedIndexes, predicate));
                }
            });
            return resultQ.take();
        } catch (InterruptedException e) {
        }
        return new ArrayList();
    }

    public Collection<Record> doQuery(Collection<Record> allRecords, Map<String, Index> namedIndexes, Predicate predicate) {
        Collection<Record> records = null;
        if (predicate != null && predicate instanceof IndexAwarePredicate) {
            List<IndexedPredicate> lsIndexPredicates = new ArrayList();
            IndexAwarePredicate iap = (IndexAwarePredicate) predicate;
            iap.collectIndexedPredicates(lsIndexPredicates);
            for (IndexedPredicate indexedPredicate : lsIndexPredicates) {
                Index index = namedIndexes.get(indexedPredicate.getIndexName());
                if (index != null) {
                    Collection<Record> sub = null;
                    if (!(indexedPredicate instanceof RangedPredicate)) {
                        sub = index.getRecords(getLongValue(indexedPredicate.getValue()));
                    } else {
                        RangedPredicate rangedPredicate = (RangedPredicate) indexedPredicate;
                        RangedPredicate.RangeType type = rangedPredicate.getRangeType();
                        if (rangedPredicate.getRangeType() == RangedPredicate.RangeType.BETWEEN) {
                            sub = index.getSubRecords(getLongValue(rangedPredicate.getFrom()), getLongValue(rangedPredicate.getTo()));
                        } else {
                            boolean equal = (type == LESS_EQUAL || type == GREATER_EQUAL);
                            if (type == LESS || type == LESS_EQUAL) {
                                sub = index.getSubRecords(equal, true, getLongValue(indexedPredicate.getValue()));
                            } else {
                                sub = index.getSubRecords(equal, false, getLongValue(indexedPredicate.getValue()));
                            }
                        }
                    }
                    logger.log(Level.FINEST, node.factory.getName() + " index sub.size " + sub.size());
                    if (records == null) {
                        records = sub;
                    } else {
                        Iterator itCurrentEntries = records.iterator();
                        while (itCurrentEntries.hasNext()) {
                            if (!sub.contains(itCurrentEntries.next())) {
                                itCurrentEntries.remove();
                            }
                        }
                    }
                }
            }
        }
        if (records == null) {
            records = allRecords;
        }
        return records;
    }


}

@@ -817,16 +817,26 @@
 ingQueue
+%3CRunnable%3E
 ();%0A%0A   
@@ -976,23 +976,16 @@
 able run
- = null
 ;%0A      
@@ -1207,32 +1207,40 @@
 ndex(final Index
+%3CRecord%3E
  index, final lo
@@ -1480,32 +1480,37 @@
 ruptedException 
+ignor
 e) %7B%0A        %7D%0A 
@@ -1546,32 +1546,40 @@
 ndex(final Index
+%3CRecord%3E
  index, final lo
@@ -1856,32 +1856,37 @@
 ruptedException 
+ignor
 e) %7B%0A        %7D%0A 
@@ -1926,24 +1926,32 @@
 (final Index
+%3CRecord%3E
  index, fina
@@ -2195,32 +2195,37 @@
 ruptedException 
+ignor
 e) %7B%0A        %7D%0A 
@@ -2634,32 +2634,40 @@
 ap%3CString, Index
+%3CRecord%3E
 %3E namedIndexes, 
@@ -2793,16 +2793,36 @@
 ingQueue
+%3CCollection%3CRecord%3E%3E
 (1);%0A   
@@ -3074,32 +3074,37 @@
 ruptedException 
+ignor
 e) %7B%0A        %7D%0A 
@@ -3122,32 +3122,40 @@
 rn new ArrayList
+%3CRecord%3E
 ();%0A    %7D%0A%0A    p
@@ -3235,16 +3235,24 @@
 g, Index
+%3CRecord%3E
 %3E namedI
@@ -3470,16 +3470,34 @@
 rrayList
+%3CIndexedPredicate%3E
 ();%0A    
@@ -3719,16 +3719,24 @@
    Index
+%3CRecord%3E
  index =
@@ -3866,23 +3866,16 @@
 ord%3E sub
- = null
 ;%0A      

clean up basic issues with generics