Add DataCommunicator for data communication of Listings

Change-Id: I1f50823fdef105c3ba0463011574908a0cec7ad9

10 files changed, 868 insertions, 7 deletions

diff --git a/client/src/main/java/com/vaadin/client/connectors/AbstractListingConnector.java b/client/src/main/java/com/vaadin/client/connectors/AbstractListingConnector.java
index 92fb0f6720..06d3ecf97d 100644
--- a/client/src/main/java/com/vaadin/client/connectors/AbstractListingConnector.java
+++ b/client/src/main/java/com/vaadin/client/connectors/AbstractListingConnector.java
@@ -15,14 +15,30 @@
  */
 package com.vaadin.client.connectors;
 
+import com.vaadin.client.connectors.data.HasDataSource;
+import com.vaadin.client.data.DataSource;
 import com.vaadin.client.ui.AbstractComponentConnector;
 import com.vaadin.ui.AbstractListing;
 
+import elemental.json.JsonObject;
+
 /**
  * Base connector class for {@link AbstractListing}.
  *
  * @since
  */
 public abstract class AbstractListingConnector
-        extends AbstractComponentConnector {
+        extends AbstractComponentConnector implements HasDataSource {
+
+    private DataSource<JsonObject> dataSource = null;
+
+    @Override
+    public void setDataSource(DataSource<JsonObject> dataSource) {
+        this.dataSource = dataSource;
+    }
+
+    @Override
+    public DataSource<JsonObject> getDataSource() {
+        return dataSource;
+    }
 }
diff --git a/client/src/main/java/com/vaadin/client/connectors/data/DataCommunicatorConnector.java b/client/src/main/java/com/vaadin/client/connectors/data/DataCommunicatorConnector.java
new file mode 100644
index 0000000000..b2b3372764
--- /dev/null
+++ b/client/src/main/java/com/vaadin/client/connectors/data/DataCommunicatorConnector.java
@@ -0,0 +1,136 @@
+/*
+ * Copyright 2000-2016 Vaadin Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package com.vaadin.client.connectors.data;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+import com.vaadin.client.ServerConnector;
+import com.vaadin.client.data.AbstractRemoteDataSource;
+import com.vaadin.client.data.DataSource;
+import com.vaadin.client.extensions.AbstractExtensionConnector;
+import com.vaadin.server.data.DataCommunicator;
+import com.vaadin.shared.data.DataCommunicatorClientRpc;
+import com.vaadin.shared.data.DataCommunicatorConstants;
+import com.vaadin.shared.data.DataRequestRpc;
+import com.vaadin.shared.ui.Connect;
+
+import elemental.json.Json;
+import elemental.json.JsonArray;
+import elemental.json.JsonObject;
+
+/**
+ * A connector for DataCommunicator class.
+ *
+ * @since
+ */
+@Connect(DataCommunicator.class)
+public class DataCommunicatorConnector extends AbstractExtensionConnector {
+
+    /**
+     * Client-side {@link DataSource} implementation to be used with
+     * {@link DataCommunicator}.
+     */
+    public class VaadinDataSource extends AbstractRemoteDataSource<JsonObject> {
+
+        private Set<String> droppedKeys = new HashSet<String>();
+
+        protected VaadinDataSource() {
+            registerRpc(DataCommunicatorClientRpc.class,
+                    new DataCommunicatorClientRpc() {
+
+                        @Override
+                        public void reset(int size) {
+                            resetDataAndSize(size);
+                        }
+
+                        @Override
+                        public void setData(int firstIndex, JsonArray data) {
+                            ArrayList<JsonObject> rows = new ArrayList<JsonObject>(
+                                    data.length());
+                            for (int i = 0; i < data.length(); i++) {
+                                JsonObject rowObject = data.getObject(i);
+                                rows.add(rowObject);
+                            }
+
+                            setRowData(firstIndex, rows);
+                        }
+
+                        @Override
+                        public void updateData(JsonArray data) {
+                            for (int i = 0; i < data.length(); ++i) {
+                                updateRowData(data.getObject(i));
+                            }
+                        }
+                    });
+        }
+
+        @Override
+        protected void requestRows(int firstRowIndex, int numberOfRows,
+                RequestRowsCallback<JsonObject> callback) {
+            getRpcProxy(DataRequestRpc.class).requestRows(firstRowIndex,
+                    numberOfRows, 0, 0);
+
+            JsonArray dropped = Json.createArray();
+            int i = 0;
+            for (String key : droppedKeys) {
+                dropped.set(i++, key);
+            }
+            droppedKeys.clear();
+
+            getRpcProxy(DataRequestRpc.class).dropRows(dropped);
+        }
+
+        @Override
+        public String getRowKey(JsonObject row) {
+            return row.getString(DataCommunicatorConstants.KEY);
+        }
+
+        @Override
+        protected void onDropFromCache(int rowIndex, JsonObject removed) {
+            droppedKeys.add(getRowKey(removed));
+
+            super.onDropFromCache(rowIndex, removed);
+        }
+
+        /**
+         * Updates row data based on row key.
+         *
+         * @param row
+         *            new row object
+         */
+        protected void updateRowData(JsonObject row) {
+            int index = indexOfKey(getRowKey(row));
+            if (index >= 0) {
+                setRowData(index, Collections.singletonList(row));
+            }
+        }
+    }
+
+    private DataSource<JsonObject> ds = new VaadinDataSource();
+
+    @Override
+    protected void extend(ServerConnector target) {
+        ServerConnector parent = getParent();
+        if (parent instanceof HasDataSource) {
+            ((HasDataSource) parent).setDataSource(ds);
+        } else {
+            assert false : "Parent not implementing HasDataSource";
+        }
+    }
+}
\ No newline at end of file
diff --git a/client/src/main/java/com/vaadin/client/connectors/data/HasDataSource.java b/client/src/main/java/com/vaadin/client/connectors/data/HasDataSource.java
new file mode 100644
index 0000000000..79a70a1fe9
--- /dev/null
+++ b/client/src/main/java/com/vaadin/client/connectors/data/HasDataSource.java
@@ -0,0 +1,41 @@
+/*
+ * Copyright 2000-2016 Vaadin Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package com.vaadin.client.connectors.data;
+
+import com.vaadin.client.data.DataSource;
+
+import elemental.json.JsonObject;
+
+/**
+ * Marker interface for Connectors that have a {@link DataSource}.
+ */
+public interface HasDataSource {
+
+    /**
+     * Sets the data source for this Connector.
+     *
+     * @param dataSource
+     *            new data source
+     */
+    void setDataSource(DataSource<JsonObject> dataSource);
+
+    /**
+     * Gets the current data source for this Connector.
+     *
+     * @return data source
+     */
+    DataSource<JsonObject> getDataSource();
+}
diff --git a/server/src/main/java/com/vaadin/server/KeyMapper.java b/server/src/main/java/com/vaadin/server/KeyMapper.java
index 35eefaff56..389fcd8b66 100644
--- a/server/src/main/java/com/vaadin/server/KeyMapper.java
+++ b/server/src/main/java/com/vaadin/server/KeyMapper.java
@@ -19,6 +19,8 @@ package com.vaadin.server;
 import java.io.Serializable;
 import java.util.HashMap;
 
+import com.vaadin.server.data.DataKeyMapper;
+
 /**
  * <code>KeyMapper</code> is the simple two-way map for generating textual keys
  * for objects and retrieving the objects later with the key.
@@ -26,7 +28,7 @@ import java.util.HashMap;
  * @author Vaadin Ltd.
  * @since 3.0
  */
-public class KeyMapper<V> implements Serializable {
+public class KeyMapper<V> implements DataKeyMapper<V>, Serializable {
 
     private int lastKey = 0;
 
diff --git a/server/src/main/java/com/vaadin/server/data/DataCommunicator.java b/server/src/main/java/com/vaadin/server/data/DataCommunicator.java
new file mode 100644
index 0000000000..8da23b42d1
--- /dev/null
+++ b/server/src/main/java/com/vaadin/server/data/DataCommunicator.java
@@ -0,0 +1,431 @@
+/*
+ * Copyright 2000-2016 Vaadin Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package com.vaadin.server.data;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import com.vaadin.server.AbstractExtension;
+import com.vaadin.server.KeyMapper;
+import com.vaadin.shared.data.DataCommunicatorClientRpc;
+import com.vaadin.shared.data.DataCommunicatorConstants;
+import com.vaadin.shared.data.DataRequestRpc;
+import com.vaadin.shared.ui.grid.Range;
+
+import elemental.json.Json;
+import elemental.json.JsonArray;
+import elemental.json.JsonObject;
+
+/**
+ * DataProvider base class. This class is the base for all DataProvider
+ * communication implementations. It uses {@link TypedDataGenerator}s to write
+ * {@link JsonObject}s representing each data object to be sent to the
+ * client-side.
+ *
+ * @since
+ */
+public class DataCommunicator<T> extends AbstractExtension {
+
+    /**
+     * Simple implementation of collection data provider communication. All data
+     * is sent by server automatically and no data is requested by client.
+     */
+    protected class SimpleDataRequestRpc implements DataRequestRpc {
+
+        @Override
+        public void requestRows(int firstRowIndex, int numberOfRows,
+                int firstCachedRowIndex, int cacheSize) {
+            pushRows = Range.withLength(firstRowIndex, numberOfRows);
+            markAsDirty();
+        }
+
+        @Override
+        public void dropRows(JsonArray keys) {
+            for (int i = 0; i < keys.length(); ++i) {
+                handler.dropActiveData(keys.getString(i));
+            }
+        }
+    }
+
+    /**
+     * A class for handling currently active data and dropping data that is no
+     * longer needed. Data tracking is based on key string provided by
+     * {@link DataKeyMapper}.
+     * <p>
+     * When the {@link DataCommunicator} is pushing new data to the client-side
+     * via {@link DataCommunicator#pushData(long, Collection)},
+     * {@link #addActiveData(Collection)} and {@link #cleanUp(Collection)} are
+     * called with the same parameter. In the clean up method any dropped data
+     * objects that are not in the given collection will be cleaned up and
+     * {@link TypedDataGenerator#destroyData(Object)} will be called for them.
+     */
+    protected class ActiveDataHandler
+            implements Serializable, TypedDataGenerator<T> {
+
+        /**
+         * Set of key strings for currently active data objects
+         */
+        private final Set<String> activeData = new HashSet<String>();
+
+        /**
+         * Set of key strings for data objects dropped on the client. This set
+         * is used to clean up old data when it's no longer needed.
+         */
+        private final Set<String> droppedData = new HashSet<String>();
+
+        /**
+         * Adds given objects as currently active objects.
+         *
+         * @param dataObjects
+         *            collection of new active data objects
+         */
+        public void addActiveData(Stream<T> dataObjects) {
+            dataObjects.map(getKeyMapper()::key)
+                    .filter(key -> !activeData.contains(key))
+                    .forEach(activeData::add);
+        }
+
+        /**
+         * Executes the data destruction for dropped data that is not sent to
+         * the client. This method takes most recently sent data objects in a
+         * collection. Doing the clean up like this prevents the
+         * {@link ActiveDataHandler} from creating new keys for rows that were
+         * dropped but got re-requested by the client-side. In the case of
+         * having all data at the client, the collection should be all the data
+         * in the back end.
+         *
+         * @param dataObjects
+         *            collection of most recently sent data to the client
+         */
+        public void cleanUp(Stream<T> dataObjects) {
+            Collection<String> keys = dataObjects.map(getKeyMapper()::key)
+                    .collect(Collectors.toSet());
+
+            // Remove still active rows that were dropped by the client
+            droppedData.removeAll(keys);
+            // Do data clean up for object no longer needed.
+            dropData(droppedData);
+            droppedData.clear();
+        }
+
+        /**
+         * Marks a data object identified by given key string to be dropped.
+         *
+         * @param key
+         *            key string
+         */
+        public void dropActiveData(String key) {
+            if (activeData.contains(key)) {
+                droppedData.add(key);
+            }
+        }
+
+        /**
+         * Returns the collection of all currently active data.
+         *
+         * @return collection of active data objects
+         */
+        public Collection<T> getActiveData() {
+            HashSet<T> hashSet = new HashSet<T>();
+            for (String key : activeData) {
+                hashSet.add(getKeyMapper().get(key));
+            }
+            return hashSet;
+        }
+
+        @Override
+        public void generateData(T data, JsonObject jsonObject) {
+            // Write the key string for given data object
+            jsonObject.put(DataCommunicatorConstants.KEY,
+                    getKeyMapper().key(data));
+        }
+
+        @Override
+        public void destroyData(T data) {
+            // Remove from active data set
+            activeData.remove(getKeyMapper().key(data));
+            // Drop the registered key
+            getKeyMapper().remove(data);
+        }
+    }
+
+    private Collection<TypedDataGenerator<T>> generators = new LinkedHashSet<TypedDataGenerator<T>>();
+    private ActiveDataHandler handler = new ActiveDataHandler();
+
+    private DataSource<T> dataSource;
+    private DataKeyMapper<T> keyMapper;
+
+    private boolean reset = false;
+    private final Set<T> updatedData = new HashSet<T>();
+    private Range pushRows = Range.withLength(0, 40);
+
+    private Comparator<T> inMemorySorting;
+    private List<SortOrder<String>> backEndSorting = new ArrayList<>();
+    private DataCommunicatorClientRpc rpc;
+
+    public DataCommunicator() {
+        addDataGenerator(handler);
+        rpc = getRpcProxy(DataCommunicatorClientRpc.class);
+        registerRpc(createRpc());
+        keyMapper = createKeyMapper();
+    }
+
+    /**
+     * Initially and in the case of a reset all data should be pushed to the
+     * client.
+     */
+    @Override
+    public void beforeClientResponse(boolean initial) {
+        super.beforeClientResponse(initial);
+
+        if (getDataSource() == null) {
+            return;
+        }
+
+        // FIXME: Sorting and Filtering with Backend
+        Set<Object> filters = Collections.emptySet();
+
+        if (initial || reset) {
+            int dataSourceSize = getDataSource().size(new Query(filters));
+            rpc.reset(dataSourceSize);
+        }
+
+        if (!pushRows.isEmpty()) {
+            int offset = pushRows.getStart();
+            int limit = pushRows.length();
+
+            Stream<T> rowsToPush;
+
+            if (getDataSource().isInMemory()) {
+                // We can safely request all the data when in memory
+                // FIXME: filter.
+                rowsToPush = getDataSource().apply(new Query());
+                if (inMemorySorting != null) {
+                    rowsToPush = rowsToPush.sorted(inMemorySorting);
+                }
+                rowsToPush = rowsToPush.skip(offset).limit(limit);
+            } else {
+                Query query = new Query(offset, limit, backEndSorting, filters);
+                rowsToPush = getDataSource().apply(query);
+            }
+            pushData(offset, rowsToPush);
+        }
+
+        if (!updatedData.isEmpty()) {
+            JsonArray dataArray = Json.createArray();
+            int i = 0;
+            for (T data : updatedData) {
+                dataArray.set(i++, getDataObject(data));
+            }
+            rpc.updateData(dataArray);
+        }
+
+        pushRows = Range.withLength(0, 0);
+        reset = false;
+        updatedData.clear();
+    }
+
+    /**
+     * Adds a {@link TypedDataGenerator} to this {@link DataCommunicator}.
+     *
+     * @param generator
+     *            typed data generator
+     */
+    public void addDataGenerator(TypedDataGenerator<T> generator) {
+        generators.add(generator);
+    }
+
+    /**
+     * Removes a {@link TypedDataGenerator} from this {@link DataCommunicator}.
+     *
+     * @param generator
+     *            typed data generator
+     */
+    public void removeDataGenerator(TypedDataGenerator<T> generator) {
+        generators.remove(generator);
+    }
+
+    /**
+     * Gets the {@link DataKeyMapper} used by this {@link DataCommunicator}. Key
+     * mapper can be used to map keys sent to the client-side back to their
+     * respective data objects.
+     *
+     * @return key mapper
+     */
+    public DataKeyMapper<T> getKeyMapper() {
+        return keyMapper;
+    }
+
+    /**
+     * Sends given collection of data objects to the client-side.
+     *
+     * @param firstIndex
+     *            first index of pushed data
+     * @param data
+     *            data objects to send as an iterable
+     */
+    protected void pushData(int firstIndex, Stream<T> data) {
+        JsonArray dataArray = Json.createArray();
+
+        int i = 0;
+        List<T> collected = data.collect(Collectors.toList());
+        for (T item : collected) {
+            dataArray.set(i++, getDataObject(item));
+        }
+
+        rpc.setData(firstIndex, dataArray);
+        handler.addActiveData(collected.stream());
+        handler.cleanUp(collected.stream());
+    }
+
+    /**
+     * Creates the JsonObject for given data object. This method calls all data
+     * generators for it.
+     *
+     * @param data
+     *            data object to be made into a json object
+     * @return json object representing the data object
+     */
+    protected JsonObject getDataObject(T data) {
+        JsonObject dataObject = Json.createObject();
+
+        for (TypedDataGenerator<T> generator : generators) {
+            generator.generateData(data, dataObject);
+        }
+
+        return dataObject;
+    }
+
+    /**
+     * Drops data objects identified by given keys from memory. This will invoke
+     * {@link TypedDataGenerator#destroyData} for each of those objects.
+     *
+     * @param droppedKeys
+     *            collection of dropped keys
+     */
+    private void dropData(Collection<String> droppedKeys) {
+        for (String key : droppedKeys) {
+            assert key != null : "Bookkeepping failure. Dropping a null key";
+
+            T data = getKeyMapper().get(key);
+            assert data != null : "Bookkeepping failure. No data object to match key";
+
+            for (TypedDataGenerator<T> g : generators) {
+                g.destroyData(data);
+            }
+        }
+    }
+
+    /**
+     * Informs the DataProvider that the collection has changed.
+     */
+    protected void reset() {
+        if (reset) {
+            return;
+        }
+
+        reset = true;
+        markAsDirty();
+    }
+
+    /**
+     * Informs the DataProvider that a data object has been updated.
+     *
+     * @param data
+     *            updated data object
+     */
+    public void refresh(T data) {
+        if (updatedData.isEmpty()) {
+            markAsDirty();
+        }
+
+        updatedData.add(data);
+    }
+
+    /**
+     * Sets the {@link Comparator} to use with in-memory sorting.
+     *
+     * @param comparator
+     *            comparator used to sort data
+     */
+    public void setInMemorySorting(Comparator<T> comparator) {
+        inMemorySorting = comparator;
+        reset();
+    }
+
+    /**
+     * Sets the {@link SortOrder}s to use with backend sorting.
+     *
+     * @param sortOrder
+     *            list of sort order information to pass to a query
+     */
+    public void setBackEndSorting(List<SortOrder<String>> sortOrder) {
+        backEndSorting.clear();
+        backEndSorting.addAll(sortOrder);
+        reset();
+    }
+
+    /**
+     * Creates a {@link DataKeyMapper} to use with this DataCommunicator.
+     * <p>
+     * This method is called from the constructor.
+     *
+     * @return key mapper
+     */
+    protected DataKeyMapper<T> createKeyMapper() {
+        return new KeyMapper<T>();
+    }
+
+    /**
+     * Creates a {@link DataRequestRpc} used with this {@link DataCommunicator}.
+     * <p>
+     * This method is called from the constructor.
+     *
+     * @return data request rpc implementation
+     */
+    protected DataRequestRpc createRpc() {
+        return new SimpleDataRequestRpc();
+    }
+
+    /**
+     * Gets the current data source from this DataCommunicator.
+     *
+     * @return the data source
+     */
+    public DataSource<T> getDataSource() {
+        return dataSource;
+    }
+
+    /**
+     * Sets the current data source for this DataCommunicator.
+     *
+     * @param dataSource
+     *            the data source to set
+     */
+    public void setDataSource(DataSource<T> dataSource) {
+        this.dataSource = dataSource;
+        reset();
+    }
+}
diff --git a/server/src/main/java/com/vaadin/server/data/DataKeyMapper.java b/server/src/main/java/com/vaadin/server/data/DataKeyMapper.java
new file mode 100644
index 0000000000..e18ea76c1e
--- /dev/null
+++ b/server/src/main/java/com/vaadin/server/data/DataKeyMapper.java
@@ -0,0 +1,62 @@
+/*
+ * Copyright 2000-2016 Vaadin Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package com.vaadin.server.data;
+
+import java.io.Serializable;
+
+/**
+ * DataKeyMapper to map data objects to key strings.
+ *
+ * @since
+ * @param <T>
+ *            data type
+ */
+public interface DataKeyMapper<T> extends Serializable {
+
+    /**
+     * Gets the key for data object. If no key exists beforehand, a new key is
+     * created.
+     *
+     * @param dataObject
+     *            data object for key mapping
+     * @return key for given data object
+     */
+    String key(T dataObject);
+
+    /**
+     * Gets the data object identified by given key.
+     *
+     * @param key
+     *            key of a data object
+     * @return identified data object; <code>null</code> if invalid key
+     */
+    T get(String key);
+
+    /**
+     * Removes a data object from the key mapping. The key is also dropped.
+     * Dropped keys are not reused.
+     *
+     * @param dataObject
+     *            dropped data object
+     */
+    void remove(T dataObject);
+
+    /**
+     * Removes all data objects from the key mapping. The keys are also dropped.
+     * Dropped keys are not reused.
+     */
+    void removeAll();
+}
\ No newline at end of file
diff --git a/server/src/main/java/com/vaadin/server/data/TypedDataGenerator.java b/server/src/main/java/com/vaadin/server/data/TypedDataGenerator.java
new file mode 100644
index 0000000000..1303921f18
--- /dev/null
+++ b/server/src/main/java/com/vaadin/server/data/TypedDataGenerator.java
@@ -0,0 +1,49 @@
+/*
+ * Copyright 2000-2016 Vaadin Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package com.vaadin.server.data;
+
+import java.io.Serializable;
+
+import elemental.json.JsonObject;
+
+/**
+ * Simple typed data generator for {@link DataCommunicator}.
+ *
+ * @since
+ */
+public interface TypedDataGenerator<T> extends Serializable {
+
+    /**
+     * Adds data for given object to {@link JsonObject}. This JsonObject will be
+     * sent to client-side DataSource.
+     *
+     * @param data
+     *            data object
+     * @param jsonObject
+     *            json object being sent to the client
+     */
+    void generateData(T data, JsonObject jsonObject);
+
+    /**
+     * Informs the {@link TypedDataGenerator} that given data has been dropped
+     * and is no longer needed. This method should clean up any unneeded
+     * information stored for this data.
+     *
+     * @param data
+     *            dropped data
+     */
+    public void destroyData(T data);
+}
\ No newline at end of file
diff --git a/server/src/main/java/com/vaadin/ui/AbstractListing.java b/server/src/main/java/com/vaadin/ui/AbstractListing.java
index 8694f5e002..64d55926d5 100644
--- a/server/src/main/java/com/vaadin/ui/AbstractListing.java
+++ b/server/src/main/java/com/vaadin/ui/AbstractListing.java
@@ -16,12 +16,13 @@
 package com.vaadin.ui;
 
 import com.vaadin.data.Listing;
+import com.vaadin.server.data.DataCommunicator;
 import com.vaadin.server.data.DataSource;
+import com.vaadin.server.data.TypedDataGenerator;
 
 /**
  * Base class for Listing components. Provides common handling for
- * {@link DataCommunicator}, {@link SelectionModel} and
- * {@link TypedDataGenerator}s.
+ * {@link DataCommunicator} and {@link TypedDataGenerator}s.
  *
  * @param <T>
  *            listing data type
@@ -29,15 +30,47 @@ import com.vaadin.server.data.DataSource;
 public abstract class AbstractListing<T> extends AbstractComponent
         implements Listing<T> {
 
-    private DataSource<T> dataSource;
+    /* DataCommunicator for this Listing component */
+    private final DataCommunicator<T> dataCommunicator = new DataCommunicator<>();
 
     @Override
     public void setDataSource(DataSource<T> dataSource) {
-        this.dataSource = dataSource;
+        getDataCommunicator().setDataSource(dataSource);
     }
 
     @Override
     public DataSource<T> getDataSource() {
-        return dataSource;
+        return getDataCommunicator().getDataSource();
+    }
+
+    /**
+     * Adds a {@link TypedDataGenerator} for the {@link DataCommunicator} of
+     * this Listing component.
+     *
+     * @param generator
+     *            typed data generator
+     */
+    protected void addDataGenerator(TypedDataGenerator<T> generator) {
+        dataCommunicator.addDataGenerator(generator);
+    }
+
+    /**
+     * Removed a {@link TypedDataGenerator} from the {@link DataCommunicator} of
+     * this Listing component.
+     *
+     * @param generator
+     *            typed data generator
+     */
+    protected void removeDataGenerator(TypedDataGenerator<T> generator) {
+        dataCommunicator.removeDataGenerator(generator);
+    }
+
+    /**
+     * Get the {@link DataCommunicator} of this Listing component.
+     *
+     * @return data provider
+     */
+    public DataCommunicator<T> getDataCommunicator() {
+        return dataCommunicator;
     }
 }
diff --git a/shared/src/main/java/com/vaadin/shared/data/DataCommunicatorClientRpc.java b/shared/src/main/java/com/vaadin/shared/data/DataCommunicatorClientRpc.java
new file mode 100644
index 0000000000..35aa64d195
--- /dev/null
+++ b/shared/src/main/java/com/vaadin/shared/data/DataCommunicatorClientRpc.java
@@ -0,0 +1,60 @@
+/*
+ * Copyright 2000-2016 Vaadin Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package com.vaadin.shared.data;
+
+import com.vaadin.shared.communication.ClientRpc;
+
+import elemental.json.JsonArray;
+
+/**
+ * RPC interface used by DataProvider to send data to the client-side.
+ *
+ * @since
+ */
+public interface DataCommunicatorClientRpc extends ClientRpc {
+
+    /**
+     * Informs the client-side DataSource that all data has been invalidated.
+     *
+     * @param size
+     *            size of the data source
+     */
+    void reset(int size);
+
+    /**
+     * Sets the data of the client-side DataSource to match the given data
+     * starting from given index.
+     * <p>
+     * <strong>Note:</strong> This method will override any existing data in the
+     * range starting from first index with the length of the data array.
+     *
+     * @param firstIndex
+     *            first index to update
+     * @param data
+     *            array of new data
+     */
+    void setData(int firstIndex, JsonArray data);
+
+    /**
+     * Updates an array of objects based on their identifying key.
+     *
+     * @param data
+     *            array of updated data
+     */
+    void updateData(JsonArray data);
+
+    // TODO: Notify add / remove
+}
\ No newline at end of file
diff --git a/shared/src/main/java/com/vaadin/shared/data/DataCommunicatorConstants.java b/shared/src/main/java/com/vaadin/shared/data/DataCommunicatorConstants.java
new file mode 100644
index 0000000000..785d04fc97
--- /dev/null
+++ b/shared/src/main/java/com/vaadin/shared/data/DataCommunicatorConstants.java
@@ -0,0 +1,31 @@
+/*
+ * Copyright 2000-2016 Vaadin Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package com.vaadin.shared.data;
+
+import java.io.Serializable;
+
+/**
+ * Set of contants used by DataCommunicator. These are commonly used JsonObject
+ * keys which are considered to be reserved for internal use.
+ *
+ * @since
+ */
+public final class DataCommunicatorConstants implements Serializable {
+    public static final String KEY = "k";
+    public static final String SELECTED = "s";
+    public static final String NAME = "n";
+    public static final String DATA = "d";
+}
\ No newline at end of file