Fix infinite loop in data communication (#8927)

* Fix infinite loop in data communication

* Switch to less aggressive reset when data communication is out of sync

* Remove triggering of reset if reset is already being performed

* Avoid redundant collection of streams in DataCommunicator

2 files changed, 16 insertions, 10 deletions

diff --git a/server/src/main/java/com/vaadin/data/provider/DataCommunicator.java b/server/src/main/java/com/vaadin/data/provider/DataCommunicator.java
index b762e3db15..566c7c732f 100644
--- a/server/src/main/java/com/vaadin/data/provider/DataCommunicator.java
+++ b/server/src/main/java/com/vaadin/data/provider/DataCommunicator.java
@@ -323,13 +323,20 @@ public class DataCommunicator<T> extends AbstractExtension {
         }
 
         Range requestedRows = getPushRows();
+        boolean triggerReset = false;
         if (!requestedRows.isEmpty()) {
             int offset = requestedRows.getStart();
             int limit = requestedRows.length();
 
             @SuppressWarnings({ "rawtypes", "unchecked" })
-            Stream<T> rowsToPush = getDataProvider().fetch(new Query(offset,
-                    limit, backEndSorting, inMemorySorting, filter));
+            List<T> rowsToPush = (List<T>) getDataProvider()
+                    .fetch(new Query(offset, limit, backEndSorting,
+                            inMemorySorting, filter))
+                    .collect(Collectors.toList());
+
+            if (!initial && !reset && rowsToPush.size() == 0) {
+                triggerReset = true;
+            }
 
             pushData(offset, rowsToPush);
         }
@@ -344,7 +351,7 @@ public class DataCommunicator<T> extends AbstractExtension {
         }
 
         setPushRows(Range.withLength(0, 0));
-        reset = false;
+        reset = triggerReset;
         updatedData.clear();
     }
 
@@ -392,18 +399,17 @@ public class DataCommunicator<T> extends AbstractExtension {
      * @param data
      *            data objects to send as an iterable
      */
-    protected void pushData(int firstIndex, Stream<T> data) {
+    protected void pushData(int firstIndex, List<T> data) {
         JsonArray dataArray = Json.createArray();
 
         int i = 0;
-        List<T> collected = data.collect(Collectors.toList());
-        for (T item : collected) {
+        for (T item : data) {
             dataArray.set(i++, getDataObject(item));
         }
 
         rpc.setData(firstIndex, dataArray);
-        handler.addActiveData(collected.stream());
-        handler.cleanUp(collected.stream());
+        handler.addActiveData(data.stream());
+        handler.cleanUp(data.stream());
     }
 
     /**
diff --git a/server/src/test/java/com/vaadin/data/provider/ReplaceDataProviderTest.java b/server/src/test/java/com/vaadin/data/provider/ReplaceDataProviderTest.java
index f64d6b27c5..fa6f832c32 100644
--- a/server/src/test/java/com/vaadin/data/provider/ReplaceDataProviderTest.java
+++ b/server/src/test/java/com/vaadin/data/provider/ReplaceDataProviderTest.java
@@ -79,7 +79,7 @@ public class ReplaceDataProviderTest {
                 beans1);
 
         dataCommunicator.setDataProvider(dataProvider, null);
-        dataCommunicator.pushData(1, beans1.stream());
+        dataCommunicator.pushData(1, beans1);
 
         SOME_BEAN bean1_17 = beans1.get(17);
         String key1_17 = dataCommunicator.getKeyMapper().key(bean1_17);
@@ -90,7 +90,7 @@ public class ReplaceDataProviderTest {
 
         dataProvider = new ListDataProvider<>(beans2);
         dataCommunicator.setDataProvider(dataProvider, null);
-        dataCommunicator.pushData(1, beans2.stream());
+        dataCommunicator.pushData(1, beans2);
 
         SOME_BEAN bean2_17 = beans2.get(17);
         String key2_17 = dataCommunicator.getKeyMapper().key(bean2_17);