ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [13/32] incubator-ignite git commit: # Renaming
Date Fri, 05 Dec 2014 10:03:14 GMT
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5fe8de26/modules/core/src/main/java/org/gridgain/grid/streamer/window/GridStreamerBoundedTimeWindow.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/streamer/window/GridStreamerBoundedTimeWindow.java b/modules/core/src/main/java/org/gridgain/grid/streamer/window/GridStreamerBoundedTimeWindow.java
deleted file mode 100644
index 34c8a54..0000000
--- a/modules/core/src/main/java/org/gridgain/grid/streamer/window/GridStreamerBoundedTimeWindow.java
+++ /dev/null
@@ -1,454 +0,0 @@
-/* @java.file.header */
-
-/*  _________        _____ __________________        _____
- *  __  ____/___________(_)______  /__  ____/______ ____(_)_______
- *  _  / __  __  ___/__  / _  __  / _  / __  _  __ `/__  / __  __ \
- *  / /_/ /  _  /    _  /  / /_/ /  / /_/ /  / /_/ / _  /  _  / / /
- *  \____/   /_/     /_/   \_,__/   \____/   \__,_/  /_/   /_/ /_/
- */
-
-package org.gridgain.grid.streamer.window;
-
-import org.gridgain.grid.*;
-import org.gridgain.grid.kernal.processors.streamer.*;
-import org.gridgain.grid.util.*;
-import org.gridgain.grid.util.typedef.internal.*;
-import org.gridgain.grid.util.lang.*;
-import org.jetbrains.annotations.*;
-
-import java.io.*;
-import java.util.*;
-import java.util.concurrent.atomic.*;
-
-/**
- * Window which is bounded by size and time interval.
- */
-public class GridStreamerBoundedTimeWindow<E> extends GridStreamerWindowAdapter<E> {
-    /** Window structures holder. */
-    private AtomicReference<WindowHolder> ref = new AtomicReference<>();
-
-    /** Time interval. */
-    private long timeInterval;
-
-    /** Window maximum size. */
-    private int maxSize;
-
-    /** Unique flag. */
-    private boolean unique;
-
-    /** Event order counter. */
-    private AtomicLong orderCnt = new AtomicLong();
-
-    /**
-     * Gets window maximum size.
-     *
-     * @return Maximum size.
-     */
-    public int getMaximumSize() {
-        return maxSize;
-    }
-
-    /**
-     * Sets window maximum size.
-     *
-     * @param maxSize Max size.
-     */
-    public void setMaximumSize(int maxSize) {
-        this.maxSize = maxSize;
-    }
-
-    /**
-     * Gets window time interval.
-     *
-     * @return Time interval.
-     */
-    public long getTimeInterval() {
-        return timeInterval;
-    }
-
-    /**
-     * Sets window time interval.
-     *
-     * @param timeInterval Time interval.
-     */
-    public void setTimeInterval(long timeInterval) {
-        this.timeInterval = timeInterval;
-    }
-
-    /**
-     * Gets window unique flag.
-     *
-     * @return {@code True} if only unique events should be added to window.
-     */
-    public boolean isUnique() {
-        return unique;
-    }
-
-    /**
-     * Sets window unique flag.
-     *
-     * @param unique {@code True} if only unique events should be added to window.
-     */
-    public void setUnique(boolean unique) {
-        this.unique = unique;
-    }
-
-    /** {@inheritDoc} */
-    @Override public void checkConfiguration() throws GridException {
-        if (timeInterval <= 0)
-            throw new GridException("Failed to initialize window (timeInterval must be positive): [windowClass=" +
-                getClass().getSimpleName() + ", maxSize=" + maxSize + ", timeInterval=" + timeInterval + ", unique=" +
-                unique + ']');
-
-        if (maxSize < 0)
-            throw new GridException("Failed to initialize window (maximumSize cannot be negative): [windowClass=" +
-                getClass().getSimpleName() + ", maxSize=" + maxSize + ", timeInterval=" + timeInterval + ", unique=" +
-                unique + ']');
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void stop0() {
-        // No-op.
-    }
-
-    /** {@inheritDoc} */
-    @SuppressWarnings("RedundantCast")
-    @Override protected void reset0() {
-        ref.set(new WindowHolder(newQueue(), unique ? (Set<Object>)new GridConcurrentHashSet<>() : null,
-            new AtomicInteger()));
-    }
-
-    /** {@inheritDoc} */
-    @Override public int size() {
-        return ref.get().size().get();
-    }
-
-    /** {@inheritDoc} */
-    @Override public int evictionQueueSize() {
-        // Get estimate for eviction queue size.
-        WindowHolder tup = ref.get();
-
-        GridConcurrentSkipListSet<Holder<E>> evtsQueue = tup.collection();
-
-        boolean sizeCheck = maxSize != 0;
-
-        int overflow = tup.size().get() - maxSize;
-
-        long timeBound = U.currentTimeMillis() - timeInterval;
-
-        int idx = 0;
-        int cnt = 0;
-
-        for (Holder holder : evtsQueue) {
-            if ((idx < overflow && sizeCheck) || holder.ts < timeBound)
-                cnt++;
-            else if ((idx >= overflow && sizeCheck) && holder.ts >= timeBound)
-                break;
-            else if (!sizeCheck && holder.ts >= timeBound)
-                break;
-
-            idx++;
-        }
-
-        return cnt;
-    }
-
-    /** {@inheritDoc} */
-    @Override protected boolean enqueue0(E evt) {
-        add(evt, U.currentTimeMillis());
-
-        return true;
-    }
-
-    /** {@inheritDoc} */
-    @Override protected Collection<E> pollEvicted0(int cnt) {
-        Collection<E> res = new ArrayList<>(cnt);
-
-        for (int i = 0; i < cnt; i++) {
-            E evicted = pollEvictedInternal();
-
-            if (evicted == null)
-                return res;
-
-            res.add(evicted);
-        }
-
-        return res;
-    }
-
-    /** {@inheritDoc} */
-    @Override protected Collection<E> pollEvictedBatch0() {
-        E res = pollEvictedInternal();
-
-        if (res == null)
-            return Collections.emptyList();
-
-        return Collections.singleton(res);
-    }
-
-    /** {@inheritDoc} */
-    @Nullable private <T> T pollEvictedInternal() {
-        WindowHolder tup = ref.get();
-
-        AtomicInteger size = tup.size();
-
-        GridConcurrentSkipListSet<Holder<E>> evtsQueue = tup.collection();
-
-        long now = U.currentTimeMillis();
-
-        while (true) {
-            int curSize = size.get();
-
-            if (maxSize > 0 && curSize > maxSize) {
-                if (size.compareAndSet(curSize, curSize - 1)) {
-                    Holder hldr = evtsQueue.pollFirst();
-
-                    if (hldr != null) {
-                        if (unique)
-                            tup.set().remove(hldr.val);
-
-                        return (T)hldr.val;
-                    }
-                    else {
-                        // No actual events in queue, it means that other thread is just adding event.
-                        // return null as it is a concurrent add call.
-                        size.incrementAndGet();
-
-                        return null;
-                    }
-                }
-            }
-            else {
-                // Check if first entry qualifies for eviction.
-                Holder first = evtsQueue.firstx();
-
-                if (first != null && first.ts < now - timeInterval) {
-                    if (evtsQueue.remove(first)) {
-                        if (unique)
-                            tup.set().remove(first.val);
-
-                        size.decrementAndGet();
-
-                        return (T)first.val;
-                    }
-                }
-                else
-                    return null;
-            }
-        }
-    }
-
-    /** {@inheritDoc} */
-    @Override protected Collection<E> dequeue0(int cnt) {
-        WindowHolder tup = ref.get();
-
-        AtomicInteger size = tup.size();
-        GridConcurrentSkipListSet<Holder<E>> evtsQueue = tup.collection();
-
-        Collection<E> resCol = new ArrayList<>(cnt);
-
-        while (true) {
-            int curSize = size.get();
-
-            if (curSize > 0) {
-                if (size.compareAndSet(curSize, curSize - 1)) {
-                    Holder<E> h = evtsQueue.pollLast();
-
-                    if (h != null) {
-                        resCol.add(h.val);
-
-                        if (unique)
-                            tup.set().remove(h.val);
-
-                        if (resCol.size() >= cnt)
-                            return resCol;
-                    }
-                    else {
-                        size.incrementAndGet();
-
-                        return resCol;
-                    }
-                }
-            }
-            else
-                return resCol;
-        }
-    }
-
-    /** {@inheritDoc} */
-    @Override protected GridStreamerWindowIterator<E> iterator0() {
-        final WindowHolder win = ref.get();
-
-        final GridConcurrentSkipListSet<Holder<E>> col = win.collection();
-        final Set<Object> set = win.set();
-
-        final Iterator<Holder<E>> it = col.iterator();
-
-        return new GridStreamerWindowIterator<E>() {
-            private Holder<E> lastRet;
-
-            @Override public boolean hasNext() {
-                return it.hasNext();
-            }
-
-            @Override public E next() {
-                lastRet = it.next();
-
-                return lastRet.val;
-            }
-
-            @Override public E removex() {
-                if (lastRet == null)
-                    throw new IllegalStateException();
-
-                if (col.remove(lastRet)) {
-                    if (set != null)
-                        set.remove(lastRet.val);
-
-                    win.size().decrementAndGet();
-
-                    return lastRet.val;
-                }
-                else
-                    return null;
-            }
-        };
-    }
-
-    /**
-     * Checks queue consistency. Used in tests.
-     */
-    void consistencyCheck() {
-        WindowHolder win = ref.get();
-
-        assert win.collection().size() == win.size().get();
-
-        if (win.set() != null) {
-            // Check no duplicates in collection.
-
-            Collection<Object> vals = new HashSet<>();
-
-            for (Object evt : win.collection())
-                assert vals.add(((Holder)evt).val);
-        }
-    }
-
-    /**
-     * @return New queue.
-     */
-    private GridConcurrentSkipListSet<Holder<E>> newQueue() {
-        return new GridConcurrentSkipListSet<>(new Comparator<Holder>() {
-            @Override public int compare(Holder h1, Holder h2) {
-                if (h1 == h2)
-                    return 0;
-
-                if (h1.ts != h2.ts)
-                    return h1.ts < h2.ts ? -1 : 1;
-
-                return h1.order < h2.order ? -1 : 1;
-            }
-        });
-    }
-
-    /**
-     * @param evt Event to add.
-     * @param ts Event timestamp.
-     */
-    private void add(E evt, long ts) {
-        WindowHolder tup = ref.get();
-
-        if (!unique) {
-            tup.collection().add(new Holder<>(evt, ts, orderCnt.incrementAndGet()));
-
-            tup.size().incrementAndGet();
-        }
-        else {
-            if (tup.set().add(evt)) {
-                tup.collection().add(new Holder<>(evt, ts, orderCnt.incrementAndGet()));
-
-                tup.size().incrementAndGet();
-            }
-        }
-    }
-
-    /**
-     * @param evts Events to add.
-     * @param ts Timestamp for added events.
-     */
-    private void addAll(Iterable<E> evts, long ts) {
-        for (E evt : evts)
-            add(evt, ts);
-    }
-
-    /**
-     * Holder.
-     */
-    private static class Holder<E> {
-        /** Value. */
-        private E val;
-
-        /** Event timestamp. */
-        private long ts;
-
-        /** Event order. */
-        private long order;
-
-        /**
-         * @param val Event.
-         * @param ts Timestamp.
-         * @param order Order.
-         */
-        private Holder(E val, long ts, long order) {
-            this.val = val;
-            this.ts = ts;
-            this.order = order;
-        }
-    }
-
-    /**
-     * Window holder.
-     */
-    @SuppressWarnings("ConstantConditions")
-    private class WindowHolder extends GridTuple3<GridConcurrentSkipListSet<Holder<E>>, Set<Object>, AtomicInteger> {
-        /** */
-        private static final long serialVersionUID = 0L;
-
-        /**
-         * Empty constructor required by {@link Externalizable}.
-         */
-        public WindowHolder() {
-            // No-op.
-        }
-
-        /**
-         * @param col Collection.
-         * @param set Set if unique.
-         * @param size Size.
-         */
-        private WindowHolder(@Nullable GridConcurrentSkipListSet<Holder<E>> col,
-            @Nullable Set<Object> set, @Nullable AtomicInteger size) {
-            super(col, set, size);
-        }
-
-        /**
-         * @return Holders collection.
-         */
-        public GridConcurrentSkipListSet<Holder<E>> collection() {
-            return get1();
-        }
-
-        /**
-         * @return Uniqueness set.
-         */
-        public Set<Object> set() {
-            return get2();
-        }
-
-        /**
-         * @return Size counter.
-         */
-        public AtomicInteger size() {
-            return get3();
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5fe8de26/modules/core/src/main/java/org/gridgain/grid/streamer/window/GridStreamerUnboundedWindow.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/streamer/window/GridStreamerUnboundedWindow.java b/modules/core/src/main/java/org/gridgain/grid/streamer/window/GridStreamerUnboundedWindow.java
deleted file mode 100644
index 9a2ceff..0000000
--- a/modules/core/src/main/java/org/gridgain/grid/streamer/window/GridStreamerUnboundedWindow.java
+++ /dev/null
@@ -1,103 +0,0 @@
-/* @java.file.header */
-
-/*  _________        _____ __________________        _____
- *  __  ____/___________(_)______  /__  ____/______ ____(_)_______
- *  _  / __  __  ___/__  / _  __  / _  / __  _  __ `/__  / __  __ \
- *  / /_/ /  _  /    _  /  / /_/ /  / /_/ /  / /_/ / _  /  _  / / /
- *  \____/   /_/     /_/   \_,__/   \____/   \__,_/  /_/   /_/ /_/
- */
-
-package org.gridgain.grid.streamer.window;
-
-import org.gridgain.grid.kernal.processors.streamer.*;
-import org.jdk8.backport.*;
-
-import java.util.*;
-
-/**
- * Unbounded window which holds all events. Events can be evicted manually from window
- * via any of the {@code dequeue(...)} methods.
- */
-public class GridStreamerUnboundedWindow<E> extends GridStreamerWindowAdapter<E> {
-    /** Events. */
-    private ConcurrentLinkedDeque8<E> evts = new ConcurrentLinkedDeque8<>();
-
-    /** {@inheritDoc} */
-    @Override protected void stop0() {
-        // No-op.
-    }
-
-    /** {@inheritDoc} */
-    @Override public void checkConfiguration() {
-        // No-op.
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void reset0() {
-        evts.clear();
-    }
-
-    /** {@inheritDoc} */
-    @Override public int size() {
-        return evts.sizex();
-    }
-
-    /** {@inheritDoc} */
-    @Override protected GridStreamerWindowIterator<E> iterator0() {
-        final ConcurrentLinkedDeque8.IteratorEx<E> it = (ConcurrentLinkedDeque8.IteratorEx<E>)evts.iterator();
-
-        return new GridStreamerWindowIterator<E>() {
-            private E lastRet;
-
-            @Override public boolean hasNext() {
-                return it.hasNext();
-            }
-
-            @Override public E next() {
-                lastRet = it.next();
-
-                return lastRet;
-            }
-
-            @Override public E removex() {
-                return (it.removex()) ? lastRet : null;
-            }
-        };
-    }
-
-    /** {@inheritDoc} */
-    @Override public int evictionQueueSize() {
-        return 0;
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean enqueue0(E evt) {
-        return evts.add(evt);
-    }
-
-    /** {@inheritDoc} */
-    @Override protected Collection<E> dequeue0(int cnt) {
-        Collection<E> res = new ArrayList<>(cnt);
-
-        for (int i = 0; i < cnt; i++) {
-            E evicted = evts.pollLast();
-
-            if (evicted != null)
-                res.add(evicted);
-            else
-                break;
-        }
-
-        return res;
-    }
-
-    /** {@inheritDoc} */
-    @Override protected Collection<E> pollEvicted0(int cnt) {
-        return Collections.emptyList();
-    }
-
-    /** {@inheritDoc} */
-    @Override protected Collection<E> pollEvictedBatch0() {
-        return Collections.emptyList();
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5fe8de26/modules/core/src/main/java/org/gridgain/grid/streamer/window/GridStreamerWindowAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/streamer/window/GridStreamerWindowAdapter.java b/modules/core/src/main/java/org/gridgain/grid/streamer/window/GridStreamerWindowAdapter.java
deleted file mode 100644
index e74d6c8..0000000
--- a/modules/core/src/main/java/org/gridgain/grid/streamer/window/GridStreamerWindowAdapter.java
+++ /dev/null
@@ -1,529 +0,0 @@
-/* @java.file.header */
-
-/*  _________        _____ __________________        _____
- *  __  ____/___________(_)______  /__  ____/______ ____(_)_______
- *  _  / __  __  ___/__  / _  __  / _  / __  _  __ `/__  / __  __ \
- *  / /_/ /  _  /    _  /  / /_/ /  / /_/ /  / /_/ / _  /  _  / / /
- *  \____/   /_/     /_/   \_,__/   \____/   \__,_/  /_/   /_/ /_/
- */
-
-package org.gridgain.grid.streamer.window;
-
-import org.apache.ignite.lang.*;
-import org.apache.ignite.lifecycle.*;
-import org.gridgain.grid.*;
-import org.gridgain.grid.kernal.processors.streamer.*;
-import org.gridgain.grid.streamer.*;
-import org.gridgain.grid.streamer.index.*;
-import org.gridgain.grid.util.typedef.*;
-import org.gridgain.grid.util.typedef.internal.*;
-import org.gridgain.grid.util.*;
-import org.jetbrains.annotations.*;
-
-import java.util.*;
-
-/**
- * Streamer window adapter.
- */
-public abstract class GridStreamerWindowAdapter<E> implements LifecycleAware, StreamerWindow<E>,
-    StreamerWindowMBean {
-    /** Default window name. */
-    private String name = getClass().getSimpleName();
-
-    /** Filter predicate. */
-    private IgnitePredicate<Object> filter;
-
-    /** Indexes. */
-    private Map<String, GridStreamerIndexProvider<E, ?, ?>> idxsAsMap;
-
-    /** */
-    private GridStreamerIndexProvider<E, ?, ?>[] idxs;
-
-    /** Lock for updates and snapshot. */
-    private final GridSpinReadWriteLock lock = new GridSpinReadWriteLock();
-
-    /** {@inheritDoc} */
-    @Override public String getClassName() {
-        return U.compact(getClass().getName());
-    }
-
-    /** {@inheritDoc} */
-    @Override public int getSize() {
-        return size();
-    }
-
-    /** {@inheritDoc} */
-    @Override public int getEvictionQueueSize() {
-        return evictionQueueSize();
-    }
-
-    /** {@inheritDoc} */
-    @Override public String name() {
-        return name;
-    }
-
-    /** {@inheritDoc} */
-    @Override public String getName() {
-        return name;
-    }
-
-    /** {@inheritDoc} */
-    @Override public Iterator<E> iterator() {
-        return new BoundedIterator(iterator0());
-    }
-
-    /**
-     * Returns an iterator over a set of elements of type T without check for iteration limit. That is,
-     * in case concurrent thread constantly adding new elements to the window we could iterate forever.
-     *
-     * @return Iterator.
-     */
-    protected abstract GridStreamerWindowIterator<E> iterator0();
-
-    /** {@inheritDoc} */
-    @Override public boolean enqueue(E evt) throws GridException {
-        lock.readLock();
-
-        try {
-            boolean res = (filter == null || filter.apply(evt));
-
-            if (res) {
-                updateIndexes(evt, false);
-
-                if (!enqueue0(evt))
-                    updateIndexes(evt, true);
-            }
-
-            return res;
-        }
-        finally {
-            lock.readUnlock();
-        }
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean enqueue(E... evts) throws GridException {
-        return enqueueAll(Arrays.asList(evts));
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean enqueueAll(Collection<E> evts) throws GridException {
-        lock.readLock();
-
-        try {
-            boolean ignoreFilter = filter == null || F.isAlwaysTrue(filter);
-
-            boolean res = true;
-
-            for (E evt : evts) {
-                if (ignoreFilter || filter.apply(evt)) {
-                    updateIndexes(evt, false);
-
-                    boolean added = enqueue0(evt);
-
-                    if (!added)
-                        updateIndexes(evt, true);
-
-                    res &= added;
-                }
-            }
-
-            return res;
-        }
-        finally {
-            lock.readUnlock();
-        }
-    }
-
-    /**
-     * Adds event to window.
-     *
-     * @param evt Event.
-     * @return {@code True} if event added.
-     */
-    protected abstract boolean enqueue0(E evt);
-
-    /** {@inheritDoc} */
-    @Override public E dequeue() throws GridException {
-        return F.first(dequeue(1));
-    }
-
-    /** {@inheritDoc} */
-    @Override public Collection<E> dequeueAll() throws GridException {
-        return dequeue(size());
-    }
-
-    /** {@inheritDoc} */
-    @Override public Collection<E> dequeue(int cnt) throws GridException {
-        lock.readLock();
-
-        try {
-            Collection<E> evts = dequeue0(cnt);
-
-            if (!evts.isEmpty() && idxs != null) {
-                for (E evt : evts)
-                    updateIndexes(evt, true);
-            }
-
-            return evts;
-        }
-        finally {
-            lock.readUnlock();
-        }
-    }
-
-    /**
-     * Dequeues up to cnt elements from window. If current window size is less than cnt, will dequeue all elements
-     * from window.
-     *
-     * @param cnt Count.
-     * @return Dequeued elements.
-     */
-    protected abstract Collection<E> dequeue0(int cnt);
-
-    /** {@inheritDoc} */
-    @Override public E pollEvicted() throws GridException {
-        return F.first(pollEvicted(1));
-    }
-
-    /** {@inheritDoc} */
-    @Override public Collection<E> pollEvictedAll() throws GridException {
-        return pollEvicted(evictionQueueSize());
-    }
-
-    /** {@inheritDoc} */
-    @Override public Collection<E> pollEvicted(int cnt) throws GridException {
-        lock.readLock();
-
-        try {
-            Collection<E> evts = pollEvicted0(cnt);
-
-            if (!evts.isEmpty() && idxs != null) {
-                for (E evt : evts)
-                    updateIndexes(evt, true);
-            }
-
-            return evts;
-        }
-        finally {
-            lock.readUnlock();
-        }
-    }
-
-    /**
-     * If window supports eviction, this method will return up to cnt evicted elements.
-     *
-     * @param cnt Count.
-     * @return Evicted elements.
-     */
-    protected abstract Collection<E> pollEvicted0(int cnt);
-
-    /** {@inheritDoc} */
-    @Override public Collection<E> pollEvictedBatch() throws GridException {
-        lock.readLock();
-
-        try {
-            Collection<E> evts = pollEvictedBatch0();
-
-            if (!evts.isEmpty() && idxs != null) {
-                for (E evt : evts)
-                    updateIndexes(evt, true);
-            }
-
-            return evts;
-        }
-        finally {
-            lock.readUnlock();
-        }
-    }
-
-    /**
-     * If window supports batch eviction, this method will poll next evicted batch from window. If windows does not
-     * support batch eviction but supports eviction, will return collection of single last evicted element. If window
-     * does not support eviction, will return empty collection.
-     *
-     * @return Elements from evicted batch.
-     */
-    protected abstract Collection<E> pollEvictedBatch0();
-
-    /** {@inheritDoc} */
-    @Override public final void start() throws GridException {
-        checkConfiguration();
-
-        if (idxs != null) {
-            for (GridStreamerIndexProvider<E, ?, ?> idx : idxs)
-                idx.initialize();
-        }
-
-        reset();
-    }
-
-    /** {@inheritDoc} */
-    @Override public final void reset(){
-        lock.writeLock();
-
-        try {
-            if (idxs != null) {
-                for (GridStreamerIndexProvider<E, ?, ?> idx : idxs)
-                    idx.reset();
-            }
-
-            reset0();
-        }
-        finally {
-            lock.writeUnlock();
-        }
-    }
-
-    /**
-     * Check window configuration.
-     *
-     * @throws GridException If failed.
-     */
-    protected abstract void checkConfiguration() throws GridException;
-
-    /**
-     * Reset routine.
-     */
-    protected abstract void reset0();
-
-    /** {@inheritDoc} */
-    @Override public void stop() {
-        lock.writeLock();
-
-        try {
-            stop0();
-        }
-        finally {
-            lock.writeUnlock();
-        }
-    }
-
-    /**
-     * Dispose window.
-     */
-    protected abstract void stop0();
-
-    /** {@inheritDoc} */
-    @Override public Collection<E> snapshot(boolean includeEvicted) {
-        lock.writeLock();
-
-        try {
-            int skip = includeEvicted ? 0 : evictionQueueSize();
-
-            List<E> res = new ArrayList<>(size() - skip);
-
-            Iterator<E> iter = iterator();
-
-            int i = 0;
-
-            while (iter.hasNext()) {
-                E next = iter.next();
-
-                if (i++ >= skip)
-                    res.add(next);
-            }
-
-            return Collections.unmodifiableList(res);
-        }
-        finally {
-            lock.writeUnlock();
-        }
-    }
-
-    /**
-     * Sets window name.
-     *
-     * @param name Window name.
-     */
-    public void setName(String name) {
-        this.name = name;
-    }
-
-    /**
-     * Gets optional event filter.
-     *
-     * @return Optional event filter.
-     */
-    @Nullable public IgnitePredicate<Object> getFilter() {
-        return filter;
-    }
-
-    /**
-     * Sets event filter.
-     *
-     * @param filter Event filter.
-     */
-    public void setFilter(@Nullable IgnitePredicate<Object> filter) {
-        this.filter = filter;
-    }
-
-    /** {@inheritDoc} */
-    @Override public <K, V> GridStreamerIndex<E, K, V> index() {
-        return index(null);
-    }
-
-    /** {@inheritDoc} */
-    @Override public <K, V> GridStreamerIndex<E, K, V> index(@Nullable String name) {
-        if (idxsAsMap != null) {
-            GridStreamerIndexProvider<E, K, V> idx = (GridStreamerIndexProvider<E, K, V>)idxsAsMap.get(name);
-
-            if (idx == null)
-                throw new IllegalArgumentException("Streamer index is not configured: " + name);
-
-            return idx.index();
-        }
-
-        throw new IllegalArgumentException("Streamer index is not configured: " + name);
-    }
-
-    /** {@inheritDoc} */
-    @Override public Collection<GridStreamerIndex<E, ?, ?>> indexes() {
-        if (idxs != null) {
-            Collection<GridStreamerIndex<E, ?, ?>> res = new ArrayList<>(idxs.length);
-
-            for (GridStreamerIndexProvider<E, ?, ?> idx : idxs)
-                res.add(idx.index());
-
-            return res;
-        }
-        else
-            return Collections.emptyList();
-    }
-
-    /**
-     * Get array of index providers.
-     *
-     * @return Index providers.
-     */
-    public GridStreamerIndexProvider<E, ?, ?>[] indexProviders() {
-        return idxs;
-    }
-
-    /**
-     * Set indexes.
-     *
-     * @param idxs Indexes.
-     * @throws IllegalArgumentException If some index names are not unique.
-     */
-    @SuppressWarnings("unchecked")
-    public void setIndexes(GridStreamerIndexProvider<E, ?, ?>... idxs) throws IllegalArgumentException {
-        A.ensure(!F.isEmpty(idxs), "!F.isEmpty(idxs)");
-
-        idxsAsMap = new HashMap<>(idxs.length, 1.0f);
-        this.idxs = new GridStreamerIndexProvider[idxs.length];
-
-        int i = 0;
-
-        for (GridStreamerIndexProvider<E, ?, ?> idx : idxs) {
-            GridStreamerIndexProvider<E, ?, ?> old = idxsAsMap.put(idx.getName(), idx);
-
-            if (old != null)
-                throw new IllegalArgumentException("Index name is not unique [idx1=" + old + ", idx2=" + idx + ']');
-
-            this.idxs[i++] = idx;
-        }
-    }
-
-    /** {@inheritDoc} */
-    @Override public void clearEvicted() throws GridException {
-        pollEvictedAll();
-    }
-
-    /**
-     * Update indexes.
-     *
-     * @param evt Event.
-     * @param rmv Remove flag.
-     * @throws GridException If index update failed.
-     */
-    protected void updateIndexes(E evt, boolean rmv) throws GridException {
-        if (idxs != null) {
-            GridStreamerIndexUpdateSync sync = new GridStreamerIndexUpdateSync();
-
-            boolean rollback = true;
-
-            try {
-                for (GridStreamerIndexProvider<E, ?, ?> idx : idxs) {
-                    if (rmv)
-                        idx.remove(sync, evt);
-                    else
-                        idx.add(sync, evt);
-                }
-
-                rollback = false;
-            }
-            finally {
-                for (GridStreamerIndexProvider<E, ?, ?> idx : idxs)
-                    idx.endUpdate(sync, evt, rollback, rmv);
-
-                sync.finish(1);
-            }
-        }
-    }
-
-    /**
-     * Window iterator wrapper which prevent returning more elements that existed in the underlying collection by the
-     * time of iterator creation.
-     */
-    private class BoundedIterator implements Iterator<E> {
-        /** Iterator. */
-        private final GridStreamerWindowIterator<E> iter;
-
-        /** How many elements to return left (at most). */
-        private int left;
-
-        /**
-         * Constructor.
-         *
-         * @param iter Iterator.
-         */
-        private BoundedIterator(GridStreamerWindowIterator<E> iter) {
-            assert iter != null;
-            assert lock != null;
-
-            this.iter = iter;
-
-            left = size();
-        }
-
-        /** {@inheritDoc} */
-        @Override public boolean hasNext() {
-            return left > 0 && iter.hasNext();
-        }
-
-        /** {@inheritDoc} */
-        @Override public E next() {
-            left--;
-
-            if (left < 0)
-                throw new NoSuchElementException();
-
-            return iter.next();
-        }
-
-        /** {@inheritDoc} */
-        @Override public void remove() {
-            if (left < 0)
-                throw new IllegalStateException();
-
-            lock.readLock();
-
-            try {
-                E evt = iter.removex();
-
-                if (evt != null) {
-                    try {
-                        updateIndexes(evt, true);
-                    }
-                    catch (GridException e) {
-                        throw new GridRuntimeException("Faied to remove event: " + evt, e);
-                     }
-                }
-            }
-            finally {
-                lock.readUnlock();
-            }
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5fe8de26/modules/core/src/main/java/org/gridgain/grid/streamer/window/StreamerBoundedSizeBatchWindow.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/streamer/window/StreamerBoundedSizeBatchWindow.java b/modules/core/src/main/java/org/gridgain/grid/streamer/window/StreamerBoundedSizeBatchWindow.java
new file mode 100644
index 0000000..c8ab1b6
--- /dev/null
+++ b/modules/core/src/main/java/org/gridgain/grid/streamer/window/StreamerBoundedSizeBatchWindow.java
@@ -0,0 +1,795 @@
+/* @java.file.header */
+
+/*  _________        _____ __________________        _____
+ *  __  ____/___________(_)______  /__  ____/______ ____(_)_______
+ *  _  / __  __  ___/__  / _  __  / _  / __  _  __ `/__  / __  __ \
+ *  / /_/ /  _  /    _  /  / /_/ /  / /_/ /  / /_/ / _  /  _  / / /
+ *  \____/   /_/     /_/   \_,__/   \____/   \__,_/  /_/   /_/ /_/
+ */
+
+package org.gridgain.grid.streamer.window;
+
+import org.gridgain.grid.*;
+import org.gridgain.grid.kernal.processors.streamer.*;
+import org.gridgain.grid.util.typedef.internal.*;
+import org.gridgain.grid.util.lang.*;
+import org.gridgain.grid.util.tostring.*;
+import org.jdk8.backport.*;
+import org.jetbrains.annotations.*;
+
+import java.io.*;
+import java.util.*;
+import java.util.concurrent.atomic.*;
+import java.util.concurrent.locks.*;
+
+/**
+ * Window that is bounded by size and accumulates events to batches.
+ */
+public class StreamerBoundedSizeBatchWindow<E> extends StreamerWindowAdapter<E> {
+    /** Max size. */
+    private int batchSize;
+
+    /** Min size. */
+    private int maxBatches;
+
+    /** Reference for queue and size. */
+    private volatile QueueHolder holder;
+
+    /** Enqueue lock. */
+    private ReadWriteLock enqueueLock = new ReentrantReadWriteLock();
+
+    /**
+     * Gets maximum number of batches can be stored in window.
+     *
+     * @return Maximum number of batches for window.
+     */
+    public int getMaximumBatches() {
+        return maxBatches;
+    }
+
+    /**
+     * Sets maximum number of batches can be stored in window.
+     *
+     * @param maxBatches Maximum number of batches for window.
+     */
+    public void setMaximumBatches(int maxBatches) {
+        this.maxBatches = maxBatches;
+    }
+
+    /**
+     * Gets batch size.
+     *
+     * @return Batch size.
+     */
+    public int getBatchSize() {
+        return batchSize;
+    }
+
+    /**
+     * Sets batch size.
+     *
+     * @param batchSize Batch size.
+     */
+    public void setBatchSize(int batchSize) {
+        this.batchSize = batchSize;
+    }
+
+    /** {@inheritDoc} */
+    @Override public void checkConfiguration() throws GridException {
+        if (batchSize <= 0)
+            throw new GridException("Failed to initialize window (batchSize size must be positive) " +
+                "[windowClass=" + getClass().getSimpleName() +
+                ", maximumBatches=" + maxBatches +
+                ", batchSize=" + batchSize + ']');
+
+        if (maxBatches < 0)
+            throw new GridException("Failed to initialize window (maximumBatches cannot be negative) " +
+                "[windowClass=" + getClass().getSimpleName() +
+                ", maximumBatches=" + maxBatches +
+                ", batchSize=" + batchSize + ']');
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void stop0() {
+        // No-op.
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void reset0() {
+        ConcurrentLinkedDeque8<Batch> first = new ConcurrentLinkedDeque8<>();
+
+        Batch b = new Batch(batchSize);
+
+        ConcurrentLinkedDeque8.Node<Batch> n = first.offerLastx(b);
+
+        b.node(n);
+
+        holder = new QueueHolder(first, new AtomicInteger(1), new AtomicInteger());
+    }
+
+    /** {@inheritDoc} */
+    @Override public int size() {
+        return holder.totalQueueSize().get();
+    }
+
+    /** {@inheritDoc} */
+    @Override protected GridStreamerWindowIterator<E> iterator0() {
+        final QueueHolder win = holder;
+
+        final Iterator<Batch> batchIt = win.batchQueue().iterator();
+
+        return new GridStreamerWindowIterator<E>() {
+            /** Current batch iterator. */
+            private ConcurrentLinkedDeque8.IteratorEx<E> curBatchIt;
+
+            /** Next batch iterator. Will be null if no more batches available. */
+            private ConcurrentLinkedDeque8.IteratorEx<E> nextBatchIt;
+
+            /** Last returned value. */
+            private E lastRet;
+
+            {
+                curBatchIt = batchIt.hasNext() ? batchIt.next().iterator() : null;
+            }
+
+            /** {@inheritDoc} */
+            @SuppressWarnings("SimplifiableIfStatement")
+            @Override public boolean hasNext() {
+                if (curBatchIt != null) {
+                    if (curBatchIt.hasNext())
+                        return true;
+
+                    return nextBatchIt != null && nextBatchIt.hasNext();
+                }
+                else
+                    return false;
+            }
+
+            /** {@inheritDoc} */
+            @Override public E next() {
+                if (curBatchIt == null)
+                    throw new NoSuchElementException();
+
+                if (!curBatchIt.hasNext()) {
+                    if (nextBatchIt != null) {
+                        curBatchIt = nextBatchIt;
+
+                        nextBatchIt = null;
+
+                        lastRet = curBatchIt.next();
+                    }
+                    else
+                        throw new NoSuchElementException();
+                }
+                else {
+                    E next = curBatchIt.next();
+
+                    // Moved to last element in batch - check for next iterator.
+                    if (!curBatchIt.hasNext())
+                        advanceBatch();
+
+                    lastRet = next;
+                }
+
+                return lastRet;
+            }
+
+            /** {@inheritDoc} */
+            @Nullable @Override public E removex() {
+                if (curBatchIt == null)
+                    throw new NoSuchElementException();
+
+                if (curBatchIt.removex()) {
+                    // Decrement global size if deleted.
+                    win.totalQueueSize().decrementAndGet();
+
+                    return lastRet;
+                }
+                else
+                    return null;
+            }
+
+            /**
+             * Moves to the next batch.
+             */
+            private void advanceBatch() {
+                if (batchIt.hasNext()) {
+                    Batch batch = batchIt.next();
+
+                    nextBatchIt = batch.iterator();
+                }
+                else
+                    nextBatchIt = null;
+            }
+        };
+    }
+
+    /** {@inheritDoc} */
+    @Override public int evictionQueueSize() {
+        QueueHolder win = holder;
+
+        int oversizeCnt = Math.max(0, win.batchQueueSize().get() - maxBatches);
+
+        Iterator<Batch> it = win.batchQueue().iterator();
+
+        int size = 0;
+
+        int idx = 0;
+
+        while (it.hasNext()) {
+            Batch batch = it.next();
+
+            if (idx++ < oversizeCnt)
+                size += batch.size();
+        }
+
+        return size;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected boolean enqueue0(E evt) {
+        try {
+            return enqueueInternal(evt);
+        }
+        catch (GridInterruptedException ignored) {
+            return false;
+        }
+    }
+
+    /**
+     * Enqueue event to window.
+     *
+     * @param evt Event to add.
+     * @return {@code True} if event was added.
+     *
+     * @throws GridInterruptedException If thread was interrupted.
+     */
+    @SuppressWarnings("LockAcquiredButNotSafelyReleased")
+    private boolean enqueueInternal(E evt) throws GridInterruptedException {
+        QueueHolder tup = holder;
+
+        ConcurrentLinkedDeque8<Batch> evts = tup.batchQueue();
+        AtomicInteger size = tup.batchQueueSize();
+
+        while (true) {
+            Batch last = evts.peekLast();
+
+            if (last == null || !last.add(evt)) {
+                // This call will ensure that last object is actually added to batch
+                // before we add new batch to events queue.
+                // If exception is thrown here, window will be left in consistent state.
+                if (last != null)
+                    last.finish();
+
+                // Add new batch to queue in write lock.
+                if (enqueueLock.writeLock().tryLock()) {
+                    try {
+                        Batch first0 = evts.peekLast();
+
+                        if (first0 == last) {
+                            Batch batch = new Batch(batchSize);
+
+                            ConcurrentLinkedDeque8.Node<Batch> node = evts.offerLastx(batch);
+
+                            batch.node(node);
+
+                            size.incrementAndGet();
+
+                            if (batch.removed() && evts.unlinkx(node))
+                                size.decrementAndGet();
+                        }
+                    }
+                    finally {
+                        enqueueLock.writeLock().unlock();
+                    }
+                }
+                else {
+                    // Acquire read lock to wait for batch enqueue.
+                    enqueueLock.readLock().lock();
+
+                    enqueueLock.readLock().unlock();
+                }
+            }
+            else {
+                // Event was added, global size increment.
+                tup.totalQueueSize().incrementAndGet();
+
+                return true;
+            }
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override protected Collection<E> pollEvicted0(int cnt) {
+        QueueHolder tup = holder;
+
+        ConcurrentLinkedDeque8<Batch> evts = tup.batchQueue();
+        AtomicInteger size = tup.batchQueueSize();
+
+        Collection<E> res = new ArrayList<>(cnt);
+
+        while (true) {
+            int curSize = size.get();
+
+            if (curSize > maxBatches) {
+                // Just peek the first batch.
+                Batch first = evts.peekFirst();
+
+                if (first != null) {
+                    assert first.finished();
+
+                    Collection<E> polled = first.pollNonBatch(cnt - res.size());
+
+                    if (!polled.isEmpty())
+                        res.addAll(polled);
+
+                    if (first.isEmpty()) {
+                        ConcurrentLinkedDeque8.Node<Batch> node = first.node();
+
+                        first.markRemoved();
+
+                        if (node != null && evts.unlinkx(node))
+                            size.decrementAndGet();
+                    }
+
+                    if (res.size() == cnt)
+                        break;
+                }
+                else
+                    break;
+            }
+            else
+                break;
+        }
+
+        // Removed entries, update global size.
+        tup.totalQueueSize().addAndGet(-res.size());
+
+        return res;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected Collection<E> pollEvictedBatch0() {
+        QueueHolder tup = holder;
+
+        ConcurrentLinkedDeque8<Batch> evts = tup.batchQueue();
+        AtomicInteger size = tup.batchQueueSize();
+
+        while (true) {
+            int curSize = size.get();
+
+            if (curSize > maxBatches) {
+                if (size.compareAndSet(curSize, curSize - 1)) {
+                    Batch polled = evts.poll();
+
+                    if (polled != null) {
+                        assert polled.finished();
+
+                        // Mark batch deleted for consistency.
+                        polled.markRemoved();
+
+                        Collection<E> polled0 = polled.shrink();
+
+                        // Result of shrink is empty, must retry the poll.
+                        if (!polled0.isEmpty()) {
+                            // Update global size.
+                            tup.totalQueueSize().addAndGet(-polled0.size());
+
+                            return polled0;
+                        }
+                    }
+                    else {
+                        // Polled was zero, so we must restore counter and return.
+                        size.incrementAndGet();
+
+                        return Collections.emptyList();
+                    }
+                }
+            }
+            else
+                return Collections.emptyList();
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override protected Collection<E> dequeue0(int cnt) {
+        QueueHolder tup = holder;
+
+        ConcurrentLinkedDeque8<Batch> evts = tup.batchQueue();
+        AtomicInteger size = tup.batchQueueSize();
+
+        Collection<E> res = new ArrayList<>(cnt);
+
+        while (true) {
+            // Just peek the first batch.
+            Batch first = evts.peekFirst();
+
+            if (first != null) {
+                Collection<E> polled = first.pollNonBatch(cnt - res.size());
+
+                // We must check for finished before unlink as no elements
+                // can be added to batch after it is finished.
+                if (first.isEmpty() && first.emptyFinished()) {
+                    ConcurrentLinkedDeque8.Node<Batch> node = first.node();
+
+                    first.markRemoved();
+
+                    if (node != null && evts.unlinkx(node))
+                        size.decrementAndGet();
+
+                    assert first.isEmpty();
+                }
+                else if (polled.isEmpty())
+                    break;
+
+                res.addAll(polled);
+
+                if (res.size() == cnt)
+                    break;
+            }
+            else
+                break;
+        }
+
+        // Update global size.
+        tup.totalQueueSize().addAndGet(-res.size());
+
+        return res;
+    }
+
+    /**
+     * Consistency check, used for testing.
+     */
+    void consistencyCheck() {
+        QueueHolder win = holder;
+
+        Iterator<E> it = iterator();
+
+        int cnt = 0;
+
+        while (it.hasNext()) {
+            it.next();
+
+            cnt++;
+        }
+
+        int cnt0 = 0;
+
+        for (Batch batch : win.batchQueue())
+            cnt0 += batch.size();
+
+        int sz = size();
+
+        assert cnt0 == sz : "Batch size comparison failed [batchCnt=" + cnt0 + ", size=" + sz + ']';
+        assert cnt == sz : "Queue size comparison failed [iterCnt=" + cnt + ", size=" + sz + ']';
+        assert win.batchQueue().size() == win.batchQueueSize().get();
+    }
+
+    /**
+     * Window structure.
+     */
+    private class QueueHolder extends GridTuple3<ConcurrentLinkedDeque8<Batch>, AtomicInteger, AtomicInteger> {
+        /** */
+        private static final long serialVersionUID = 0L;
+
+        /**
+         * Empty constructor required by {@link Externalizable}.
+         */
+        public QueueHolder() {
+            // No-op.
+        }
+
+        /**
+         * @param batchQueue Batch queue.
+         * @param batchQueueSize Batch queue size counter.
+         * @param globalSize Global size counter.
+         */
+        private QueueHolder(ConcurrentLinkedDeque8<Batch> batchQueue,
+            AtomicInteger batchQueueSize, @Nullable AtomicInteger globalSize) {
+            super(batchQueue, batchQueueSize, globalSize);
+
+            assert batchQueue.size() == 1;
+            assert batchQueueSize.get() == 1;
+        }
+
+        /**
+         * @return Events queue.
+         */
+        @SuppressWarnings("ConstantConditions")
+        public ConcurrentLinkedDeque8<Batch> batchQueue() {
+            return get1();
+        }
+
+        /**
+         * @return Batch queue size.
+         */
+        @SuppressWarnings("ConstantConditions")
+        public AtomicInteger batchQueueSize() {
+            return get2();
+        }
+
+        /**
+         * @return Global queue size.
+         */
+        @SuppressWarnings("ConstantConditions")
+        public AtomicInteger totalQueueSize() {
+            return get3();
+        }
+    }
+
+    /**
+     * Batch.
+     */
+    private class Batch extends ReentrantReadWriteLock implements Iterable<E> {
+        /** */
+        private static final long serialVersionUID = 0L;
+
+        /** Batch events. */
+        private ConcurrentLinkedDeque8<E> evts;
+
+        /** Capacity. */
+        private AtomicInteger cap;
+
+        /** Finished. */
+        private volatile boolean finished;
+
+        /** Queue node. */
+        @GridToStringExclude
+        private ConcurrentLinkedDeque8.Node<Batch> qNode;
+
+        /** Node removed flag. */
+        private volatile boolean rmvd;
+
+        /**
+         * @param batchSize Batch size.
+         */
+        private Batch(int batchSize) {
+            cap = new AtomicInteger(batchSize);
+
+            evts = new ConcurrentLinkedDeque8<>();
+        }
+
+        /**
+         * @return {@code True} if batch is removed.
+         */
+        public boolean removed() {
+            return rmvd;
+        }
+
+        /**
+         * Marks batch as removed.
+         */
+        public void markRemoved() {
+            rmvd = true;
+        }
+
+        /**
+         * Adds event to batch.
+         *
+         * @param evt Event to add.
+         * @return {@code True} if event was added, {@code false} if batch is full.
+         */
+        public boolean add(E evt) {
+            readLock().lock();
+
+            try {
+                if (finished)
+                    return false;
+
+                while (true) {
+                    int size = cap.get();
+
+                    if (size > 0) {
+                        if (cap.compareAndSet(size, size - 1)) {
+                            evts.add(evt);
+
+                            // Will go through write lock and finish batch.
+                            if (size == 1)
+                                finished = true;
+
+                            return true;
+                        }
+                    }
+                    else
+                        return false;
+                }
+            }
+            finally {
+                readLock().unlock();
+            }
+        }
+
+        /**
+         * @return Queue node.
+         */
+        public ConcurrentLinkedDeque8.Node<Batch> node() {
+            return qNode;
+        }
+
+        /**
+         * @param qNode Queue node.
+         */
+        public void node(ConcurrentLinkedDeque8.Node<Batch> qNode) {
+            this.qNode = qNode;
+        }
+
+        /**
+         * Waits for latch count down after last event was added.
+         *
+         * @throws GridInterruptedException If wait was interrupted.
+         */
+        public void finish() throws GridInterruptedException {
+            writeLock().lock();
+
+            try {
+                // Safety.
+                assert cap.get() == 0;
+                assert finished;
+            }
+            finally {
+                writeLock().unlock();
+            }
+        }
+
+        /**
+         * @return {@code True} if batch is finished and no more events will be added to it.
+         */
+        public boolean finished() {
+            readLock().lock();
+
+            try {
+                return finished;
+            }
+            finally {
+                readLock().unlock();
+            }
+        }
+
+        /**
+         * Gets batch size.
+         *
+         * @return Batch size.
+         */
+        public int size() {
+            readLock().lock();
+
+            try {
+                return evts == null ? 0 : evts.sizex();
+            }
+            finally {
+                readLock().unlock();
+            }
+        }
+
+        /**
+         * @return {@code True} if batch is empty.
+         */
+        public boolean isEmpty() {
+            readLock().lock();
+
+            try {
+                return evts == null || evts.isEmpty();
+            }
+            finally {
+                readLock().unlock();
+            }
+        }
+
+        /**
+         * Checks if batch is empty and finished inside write lock. This will ensure that no more entries will
+         * be added to batch and it can be safely unlinked from the queue.
+         *
+         * @return {@code True} if batch is empty and finished.
+         */
+        public boolean emptyFinished() {
+            writeLock().lock();
+
+            try {
+                return finished && (evts == null || evts.isEmpty());
+            }
+            finally {
+                writeLock().unlock();
+            }
+        }
+
+        /** {@inheritDoc} */
+        @Override public ConcurrentLinkedDeque8.IteratorEx<E> iterator() {
+            readLock().lock();
+
+            try {
+                if (evts != null)
+                    return (ConcurrentLinkedDeque8.IteratorEx<E>)evts.iterator();
+
+                return new ConcurrentLinkedDeque8.IteratorEx<E>() {
+                    @Override public boolean removex() {
+                        throw new NoSuchElementException();
+                    }
+
+                    @Override public boolean hasNext() {
+                        return false;
+                    }
+
+                    @Override public E next() {
+                        throw new NoSuchElementException();
+                    }
+
+                    @Override public void remove() {
+                        throw new NoSuchElementException();
+                    }
+                };
+            }
+            finally {
+                readLock().unlock();
+            }
+        }
+
+        /**
+         * Polls up to {@code cnt} objects from batch in concurrent fashion.
+         *
+         * @param cnt Number of objects to poll.
+         * @return Collection of polled elements or empty collection if nothing to poll.
+         */
+        public Collection<E> pollNonBatch(int cnt) {
+            readLock().lock();
+
+            try {
+                if (evts == null)
+                    return Collections.emptyList();
+
+                Collection<E> res = new ArrayList<>(cnt);
+
+                for (int i = 0; i < cnt; i++) {
+                    E evt = evts.poll();
+
+                    if (evt != null)
+                        res.add(evt);
+                    else
+                        return res;
+                }
+
+                return res;
+            }
+            finally {
+                readLock().unlock();
+            }
+        }
+
+        /**
+         * Shrinks this batch. No events can be polled from it after this method.
+         *
+         * @return Collection of events contained in batch before shrink (empty collection in
+         *         case no events were present).
+         */
+        public Collection<E> shrink() {
+            writeLock().lock();
+
+            try {
+                if (evts == null)
+                    return Collections.emptyList();
+
+                // Since iterator can concurrently delete elements, we must poll here.
+                Collection<E> res = new ArrayList<>(evts.sizex());
+
+                E o;
+
+                while ((o = evts.poll()) != null)
+                    res.add(o);
+
+                // Nothing cal be polled after shrink.
+                evts = null;
+
+                return res;
+            }
+            finally {
+                writeLock().unlock();
+            }
+        }
+
+        /** {@inheritDoc} */
+        @Override public String toString() {
+            ConcurrentLinkedDeque8<E> evts0 = evts;
+
+            return S.toString(Batch.class, this, "evtQueueSize", evts0 == null ? 0 : evts0.sizex());
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5fe8de26/modules/core/src/main/java/org/gridgain/grid/streamer/window/StreamerBoundedSizeSortedWindow.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/streamer/window/StreamerBoundedSizeSortedWindow.java b/modules/core/src/main/java/org/gridgain/grid/streamer/window/StreamerBoundedSizeSortedWindow.java
new file mode 100644
index 0000000..c797976
--- /dev/null
+++ b/modules/core/src/main/java/org/gridgain/grid/streamer/window/StreamerBoundedSizeSortedWindow.java
@@ -0,0 +1,210 @@
+/* @java.file.header */
+
+/*  _________        _____ __________________        _____
+ *  __  ____/___________(_)______  /__  ____/______ ____(_)_______
+ *  _  / __  __  ___/__  / _  __  / _  / __  _  __ `/__  / __  __ \
+ *  / /_/ /  _  /    _  /  / /_/ /  / /_/ /  / /_/ / _  /  _  / / /
+ *  \____/   /_/     /_/   \_,__/   \____/   \__,_/  /_/   /_/ /_/
+ */
+
+package org.gridgain.grid.streamer.window;
+
+import org.gridgain.grid.kernal.processors.streamer.*;
+import org.gridgain.grid.util.*;
+import org.gridgain.grid.util.typedef.*;
+import org.jetbrains.annotations.*;
+
+import java.util.*;
+import java.util.concurrent.atomic.*;
+
+/**
+ * Size-bounded sorted window. Unlike {@link StreamerBoundedSizeWindow}, which limits
+ * window only on size, this window also provides events in sorted order.
+ */
+public class StreamerBoundedSizeSortedWindow<E>
+    extends StreamerBoundedSizeWindowAdapter<E, StreamerBoundedSizeSortedWindow.Holder<E>> {
+    /** Comparator. */
+    private Comparator<E> comp;
+
+    /** Order counter. */
+    private AtomicLong orderCnt = new AtomicLong();
+
+    /**
+     * Gets event comparator.
+     *
+     * @return Event comparator.
+     */
+    public Comparator<E> getComparator() {
+        return comp;
+    }
+
+    /**
+     * Sets event comparator.
+     *
+     * @param comp Comparator.
+     */
+    public void setComparator(Comparator<E> comp) {
+        this.comp = comp;
+    }
+
+    /** {@inheritDoc} */
+    @SuppressWarnings("unchecked")
+    @Override protected Collection<Holder<E>> newCollection() {
+        final Comparator<E> comp0 = comp;
+
+        Collection<Holder<E>> col = new GridConcurrentSkipListSet<>(new Comparator<Holder<E>>() {
+            @Override public int compare(Holder<E> h1, Holder<E> h2) {
+                if (h1 == h2)
+                    return 0;
+
+                int diff = comp0 == null ?
+                    ((Comparable<E>)h1.val).compareTo(h2.val) : comp0.compare(h1.val, h2.val);
+
+                if (diff != 0)
+                    return diff;
+                else {
+                    assert h1.order != h2.order;
+
+                    return h1.order < h2.order ? -1 : 1;
+                }
+            }
+        });
+
+        return (Collection)col;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected boolean addInternal(E evt, Collection<Holder<E>> col, @Nullable Set<E> set) {
+        if (comp == null) {
+            if (!(evt instanceof Comparable))
+                throw new IllegalArgumentException("Failed to add object to window (object is not comparable and no " +
+                    "comparator is specified: " + evt);
+        }
+
+        if (set != null) {
+            if (set.add(evt)) {
+                col.add(new Holder<>(evt, orderCnt.getAndIncrement()));
+
+                return true;
+            }
+
+            return false;
+        }
+        else {
+            col.add(new Holder<>(evt, orderCnt.getAndIncrement()));
+
+            return true;
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override protected int addAllInternal(Collection<E> evts, Collection<Holder<E>> col, @Nullable Set<E> set) {
+        int cnt = 0;
+
+        for (E evt : evts) {
+            if (addInternal(evt, col, set))
+                cnt++;
+        }
+
+        return cnt;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected E pollInternal(Collection<Holder<E>> col, Set<E> set) {
+        Holder<E> h = (Holder<E>)((NavigableSet<E>)col).pollLast();
+
+        if (set != null && h != null)
+            set.remove(h.val);
+
+        return h == null ? null : h.val;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected GridStreamerWindowIterator<E> iteratorInternal(final Collection<Holder<E>> col,
+        final Set<E> set, final AtomicInteger size) {
+        final Iterator<Holder<E>> it = col.iterator();
+
+        return new GridStreamerWindowIterator<E>() {
+            private Holder<E> lastRet;
+
+            @Override public boolean hasNext() {
+                return it.hasNext();
+            }
+
+            @Override public E next() {
+                lastRet = it.next();
+
+                return lastRet.val;
+            }
+
+            @Override public E removex() {
+                if (lastRet == null)
+                    throw new IllegalStateException();
+
+                if (col.remove(lastRet)) {
+                    if (set != null)
+                        set.remove(lastRet.val);
+
+                    size.decrementAndGet();
+
+                    return lastRet.val;
+                }
+                else
+                    return null;
+            }
+        };
+    }
+
+    /**
+     * Value wrapper.
+     */
+    @SuppressWarnings("PackageVisibleInnerClass")
+    static class Holder<E> {
+        /** Value. */
+        private E val;
+
+        /** Order to distinguish between objects for which comparator returns 0. */
+        private long order;
+
+        /**
+         * @param val Value to hold.
+         * @param order Adding order.
+         */
+        private Holder(E val, long order) {
+            this.val = val;
+            this.order = order;
+        }
+
+        /** {@inheritDoc} */
+        @Override public int hashCode() {
+            return val.hashCode();
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean equals(Object obj) {
+            if (this == obj)
+                return false;
+
+            if (!(obj instanceof Holder))
+                return false;
+
+            Holder h = (Holder)obj;
+
+            return F.eq(val, h.val) && order == h.order;
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void consistencyCheck(Collection<Holder<E>> col, Set<E> set, AtomicInteger size) {
+        assert col.size() == size.get();
+
+        if (set != null) {
+            // Check no duplicates in collection.
+
+            Collection<Object> vals = new HashSet<>();
+
+            for (Object evt : col)
+                assert vals.add(((Holder)evt).val);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5fe8de26/modules/core/src/main/java/org/gridgain/grid/streamer/window/StreamerBoundedSizeWindow.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/streamer/window/StreamerBoundedSizeWindow.java b/modules/core/src/main/java/org/gridgain/grid/streamer/window/StreamerBoundedSizeWindow.java
new file mode 100644
index 0000000..ff6fedc
--- /dev/null
+++ b/modules/core/src/main/java/org/gridgain/grid/streamer/window/StreamerBoundedSizeWindow.java
@@ -0,0 +1,136 @@
+/* @java.file.header */
+
+/*  _________        _____ __________________        _____
+ *  __  ____/___________(_)______  /__  ____/______ ____(_)_______
+ *  _  / __  __  ___/__  / _  __  / _  / __  _  __ `/__  / __  __ \
+ *  / /_/ /  _  /    _  /  / /_/ /  / /_/ /  / /_/ / _  /  _  / / /
+ *  \____/   /_/     /_/   \_,__/   \____/   \__,_/  /_/   /_/ /_/
+ */
+
+package org.gridgain.grid.streamer.window;
+
+import org.gridgain.grid.kernal.processors.streamer.*;
+import org.jdk8.backport.*;
+import org.jetbrains.annotations.*;
+
+import java.util.*;
+import java.util.concurrent.atomic.*;
+
+/**
+ * Queue window bounded by number of elements in queue. After adding elements to this window called
+ * must check for evicted events.
+ * <p>
+ * It is guaranteed that window size will never get less then maximum size when poling from this window
+ * concurrently from different threads.
+ */
+public class StreamerBoundedSizeWindow<E> extends StreamerBoundedSizeWindowAdapter<E, E> {
+    /** {@inheritDoc} */
+    @Override protected Collection<E> newCollection() {
+        return new ConcurrentLinkedDeque8<>();
+    }
+
+    /** {@inheritDoc} */
+    @Override public GridStreamerWindowIterator<E> iteratorInternal(Collection<E> col, final Set<E> set,
+        final AtomicInteger size) {
+        final ConcurrentLinkedDeque8.IteratorEx<E> it =
+            (ConcurrentLinkedDeque8.IteratorEx<E>)col.iterator();
+
+        return new GridStreamerWindowIterator<E>() {
+            private E lastRet;
+
+            @Override public boolean hasNext() {
+                return it.hasNext();
+            }
+
+            @Override public E next() {
+                lastRet = it.next();
+
+                return lastRet;
+            }
+
+            @Override public E removex() {
+                if (it.removex()) {
+                    if (set != null)
+                        set.remove(lastRet);
+
+                    size.decrementAndGet();
+
+                    return lastRet;
+                }
+                else
+                    return null;
+            }
+        };
+    }
+
+    /** {@inheritDoc} */
+    @SuppressWarnings("IfMayBeConditional")
+    @Override protected boolean addInternal(E evt, Collection<E> col, Set<E> set) {
+        assert col instanceof ConcurrentLinkedDeque8;
+
+        // If unique.
+        if (set != null) {
+            if (set.add(evt)) {
+                col.add(evt);
+
+                return true;
+            }
+
+            return false;
+        }
+        else {
+            col.add(evt);
+
+            return true;
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override protected int addAllInternal(Collection<E> evts, Collection<E> col, Set<E> set) {
+        assert col instanceof ConcurrentLinkedDeque8;
+        if (set != null) {
+            int cnt = 0;
+
+            for (E evt : evts) {
+                if (set.add(evt)) {
+                    col.add(evt);
+
+                    cnt++;
+                }
+            }
+
+            return cnt;
+        }
+        else {
+            col.addAll(evts);
+
+            return evts.size();
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Nullable @Override protected E pollInternal(Collection<E> col, Set<E> set) {
+        assert col instanceof ConcurrentLinkedDeque8;
+
+        E res = ((Queue<E>)col).poll();
+
+        if (set != null && res != null)
+            set.remove(res);
+
+        return res;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void consistencyCheck(Collection<E> col, Set<E> set, AtomicInteger size) {
+        assert col.size() == size.get();
+
+        if (set != null) {
+            // Check no duplicates in collection.
+
+            Collection<Object> vals = new HashSet<>();
+
+            for (Object evt : col)
+                assert vals.add(evt);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5fe8de26/modules/core/src/main/java/org/gridgain/grid/streamer/window/StreamerBoundedSizeWindowAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/streamer/window/StreamerBoundedSizeWindowAdapter.java b/modules/core/src/main/java/org/gridgain/grid/streamer/window/StreamerBoundedSizeWindowAdapter.java
new file mode 100644
index 0000000..57110f7
--- /dev/null
+++ b/modules/core/src/main/java/org/gridgain/grid/streamer/window/StreamerBoundedSizeWindowAdapter.java
@@ -0,0 +1,349 @@
+/* @java.file.header */
+
+/*  _________        _____ __________________        _____
+ *  __  ____/___________(_)______  /__  ____/______ ____(_)_______
+ *  _  / __  __  ___/__  / _  __  / _  / __  _  __ `/__  / __  __ \
+ *  / /_/ /  _  /    _  /  / /_/ /  / /_/ /  / /_/ / _  /  _  / / /
+ *  \____/   /_/     /_/   \_,__/   \____/   \__,_/  /_/   /_/ /_/
+ */
+
+package org.gridgain.grid.streamer.window;
+
+import org.gridgain.grid.*;
+import org.gridgain.grid.kernal.processors.streamer.*;
+import org.gridgain.grid.util.*;
+import org.gridgain.grid.util.lang.*;
+import org.jetbrains.annotations.*;
+
+import java.io.*;
+import java.util.*;
+import java.util.concurrent.atomic.*;
+
+/**
+ * Abstract non-public class for size-bound windows. Support reset.
+ */
+abstract class StreamerBoundedSizeWindowAdapter<E, T> extends StreamerWindowAdapter<E> {
+    /** Reference. */
+    private AtomicReference<WindowHolder> ref = new AtomicReference<>();
+
+    /** If true, only unique elements will be accepted. */
+    private boolean unique;
+
+    /** Window maximum size. */
+    protected int maxSize;
+
+    /**
+     * Gets window maximum size.
+     *
+     * @return Maximum size.
+     */
+    public int getMaximumSize() {
+        return maxSize;
+    }
+
+    /**
+     * Sets window maximum size.
+     *
+     * @param maxSize Maximum size.
+     */
+    public void setMaximumSize(int maxSize) {
+        this.maxSize = maxSize;
+    }
+
+    /**
+     * @return True if only unique elements will be accepted.
+     */
+    public boolean isUnique() {
+        return unique;
+    }
+
+    /**
+     * @param unique If true, only unique elements will be accepted.
+     */
+    public void setUnique(boolean unique) {
+        this.unique = unique;
+    }
+
+    /** {@inheritDoc} */
+    @Override public void checkConfiguration() throws GridException {
+        if (maxSize < 0)
+            throw new GridException("Failed to initialize window (maximumSize cannot be negative) " +
+                "[windowClass=" + getClass().getSimpleName() +
+                ", maxSize=" + maxSize +
+                ", unique=" + unique + ']');
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void stop0() {
+        // No-op.
+    }
+
+    /** {@inheritDoc} */
+    @Override public int size() {
+        int size = ref.get().size().get();
+
+        return size > 0 ? size : 0;
+    }
+
+    /** {@inheritDoc} */
+    @Override public int evictionQueueSize() {
+        int evictSize = size() - maxSize;
+
+        return evictSize > 0 ? evictSize : 0;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected boolean enqueue0(E evt) {
+        add(evt);
+
+        return true;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected Collection<E> pollEvicted0(int cnt) {
+        Collection<E> res = new ArrayList<>(cnt);
+
+        for (int i = 0; i < cnt; i++) {
+            E evicted = pollEvictedInternal();
+
+            if (evicted == null)
+                return res;
+
+            res.add(evicted);
+        }
+
+        return res;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected Collection<E> pollEvictedBatch0() {
+        E res = pollEvictedInternal();
+
+        if (res == null)
+            return Collections.emptyList();
+
+        return Collections.singleton(res);
+    }
+
+    /**
+     * Poll evicted internal implementation.
+     *
+     * @return Evicted element.
+     */
+    @Nullable private E pollEvictedInternal() {
+        WindowHolder tup = ref.get();
+
+        AtomicInteger size = tup.size();
+
+        while (true) {
+            int curSize = size.get();
+
+            if (curSize > maxSize) {
+                if (size.compareAndSet(curSize, curSize - 1)) {
+                    E evt = pollInternal(tup.collection(), tup.set());
+
+                    if (evt != null)
+                        return evt;
+                    else {
+                        // No actual events in queue, it means that other thread is just adding event.
+                        // return null as it is a concurrent add call.
+                        size.incrementAndGet();
+
+                        return null;
+                    }
+                }
+            }
+            else
+                return null;
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override protected Collection<E> dequeue0(int cnt) {
+        WindowHolder tup = ref.get();
+
+        AtomicInteger size = tup.size();
+        Collection<T> evts = tup.collection();
+
+        Collection<E> resCol = new ArrayList<>(cnt);
+
+        while (true) {
+            int curSize = size.get();
+
+            if (curSize > 0) {
+                if (size.compareAndSet(curSize, curSize - 1)) {
+                    E res = pollInternal(evts, tup.set());
+
+                    if (res != null) {
+                        resCol.add(res);
+
+                        if (resCol.size() >= cnt)
+                            return resCol;
+                    }
+                    else {
+                        size.incrementAndGet();
+
+                        return resCol;
+                    }
+                }
+            }
+            else
+                return resCol;
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override protected GridStreamerWindowIterator<E> iterator0() {
+        WindowHolder win = ref.get();
+
+        return iteratorInternal(win.collection(), win.set(), win.size());
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void reset0() {
+        ref.set(new WindowHolder(newCollection(),
+            unique ? new GridConcurrentHashSet<E>() : null,
+            new AtomicInteger()));
+    }
+
+    /**
+     * @param evt Event to add.
+     */
+    private void add(E evt) {
+        WindowHolder tup = ref.get();
+
+        if (addInternal(evt, tup.collection(), tup.set()))
+            tup.size().incrementAndGet();
+    }
+
+    /**
+     * @param evts Events to add.
+     */
+    private void addAll(Collection<E> evts) {
+        WindowHolder tup = ref.get();
+
+        int cnt = addAllInternal(evts, tup.collection(), tup.set());
+
+        tup.size().addAndGet(cnt);
+    }
+
+    /**
+     * Checks window consistency. Used for testing.
+     */
+    void consistencyCheck() {
+        WindowHolder win = ref.get();
+
+        consistencyCheck(win.collection(), win.set(), win.size());
+    }
+
+    /**
+     * Get underlying collection.
+     *
+     * @return Collection.
+     */
+    @SuppressWarnings("ConstantConditions")
+    protected Collection<T> collection() {
+        return ref.get().get1();
+    }
+
+    /**
+     * Creates new collection specific for window implementation. This collection will be subsequently passed
+     * to addInternal(...) and pollInternal() methods.
+     *
+     * @return Collection - holder.
+     */
+    protected abstract Collection<T> newCollection();
+
+    /**
+     * Adds event to queue implementation.
+     *
+     * @param evt Event to add.
+     * @param col Collection to add to.
+     * @param set Set to check.
+     * @return {@code True} if event was added.
+     */
+    protected abstract boolean addInternal(E evt, Collection<T> col, @Nullable Set<E> set);
+
+    /**
+     * Adds all events to queue implementation.
+     *
+     * @param evts Events to add.
+     * @param col Collection to add to.
+     * @param set Set to check.
+     * @return Added events number.
+     */
+    protected abstract int addAllInternal(Collection<E> evts, Collection<T> col, @Nullable Set<E> set);
+
+    /**
+     * @param col Collection to add to.
+     * @param set Set to check.
+     * @return Polled object.
+     */
+    @Nullable protected abstract E pollInternal(Collection<T> col, @Nullable Set<E> set);
+
+    /**
+     * Creates iterator based on implementation collection type.
+     *
+     * @param col Collection.
+     * @param set Set to check.
+     * @param size Size.
+     * @return Iterator.
+     */
+    protected abstract GridStreamerWindowIterator<E> iteratorInternal(Collection<T> col, @Nullable Set<E> set,
+        AtomicInteger size);
+
+    /**
+     * Checks consistency. Used in tests.
+     *
+     * @param col Collection.
+     * @param set Set if unique.
+     * @param size Size holder.
+     */
+    protected abstract void consistencyCheck(Collection<T> col, Set<E> set, AtomicInteger size);
+
+    /**
+     * Window holder.
+     */
+    @SuppressWarnings("ConstantConditions")
+    private class WindowHolder extends GridTuple3<Collection<T>, Set<E>, AtomicInteger> {
+        /** */
+        private static final long serialVersionUID = 0L;
+
+        /**
+         * Empty constructor required by {@link Externalizable}.
+         */
+        public WindowHolder() {
+            // No-op.
+        }
+
+        /**
+         * @param col Collection.
+         * @param set Set if unique.
+         * @param size Window size counter.
+         */
+        WindowHolder(@Nullable Collection<T> col, @Nullable Set<E> set, @Nullable AtomicInteger size) {
+            super(col, set, size);
+        }
+
+        /**
+         * @return Collection.
+         */
+        public Collection<T> collection() {
+            return get1();
+        }
+
+        /**
+         * @return Set.
+         */
+        public Set<E> set() {
+            return get2();
+        }
+
+        /**
+         * @return Size.
+         */
+        public AtomicInteger size() {
+            return get3();
+        }
+    }
+}


Mime
View raw message