feat: Implement Batch Streaming & Progressive Rendering (#9094)

Author: tobiu

src/data/Store.mjs

@@ -416,13 +416,6 @@ class Store extends Collection {
         return ClassSystemUtil.beforeSetInstance(value);
     }
 
-    /**
-     * @param {Object} data
-     */
-    onProxyData(data) {
-        this.add(data);
-    }
-
     /**
      * @param {Object[]|Neo.data.Model[]} value
      * @param {Object[]|Neo.data.Model[]} oldValue
@@ -752,10 +745,21 @@ class Store extends Collection {
                 me.clear();
             }
 
-            me.isLoading     = true;
-            me.suspendEvents = true;
+            me.isLoading = true;
 
-            const onData = me.onProxyData.bind(me);
+            const onData = (data) => {
+                me.add(data);
+                
+                // Progressive Rendering:
+                // As soon as we have data, we want the grid to render.
+                if (me.isLoading) {
+                    me.isLoading = false;
+                }
+                
+                // Fire load event to trigger grid updates (row count, scrollbar)
+                // GridBody checks data.total or store.count.
+                me.fire('load', {items: me.items, total: me.count});
+            };
 
             me.proxy.on('data', onData);
 
@@ -768,12 +772,11 @@ class Store extends Collection {
                 const response = await me.proxy.read(params);
 
                 me.proxy.un('data', onData);
-                me.suspendEvents = false;
 
                 if (response.success) {
                     me.totalCount = response.totalCount || me.count;
                     me.isLoaded   = true;
-                    me.isLoading  = false;
+                    me.isLoading  = false; // Ensure it's false at the end
                     me.fire('load', {items: me.items, total: me.totalCount});
                     return me.items;
                 } else {
@@ -782,8 +785,7 @@ class Store extends Collection {
                 }
             } catch (e) {
                 me.proxy.un('data', onData);
-                me.suspendEvents = false;
-                me.isLoading     = false;
+                me.isLoading = false;
                 throw e;
             }
         } else {

src/data/proxy/Stream.mjs

@@ -15,17 +15,24 @@ class Stream extends Base {
          * @member {String} ntype='proxy-stream'
          * @protected
          */
-        ntype: 'proxy-stream'
+        ntype: 'proxy-stream',
+        /**
+         * Number of records to batch before firing a 'data' event.
+         * @member {Number} chunkSize=500
+         */
+        chunkSize: 500
     }
 
     /**
      * @param {Object} operation
      * @returns {Promise}
      */
     async read(operation) {
-        let me  = this,
-            url = me.url || operation.url,
-            count = 0;
+        let me        = this,
+            chunk     = [],
+            chunkSize = me.chunkSize,
+            count     = 0,
+            url       = me.url || operation.url;
 
         if (!url) {
             throw new Error('No URL specified');
@@ -53,9 +60,13 @@ class Stream extends Base {
             if (done) {
                 // Process any remaining buffer
                 if (buffer.trim()) {
-                    me.processLine(buffer);
+                    me.processLine(buffer, chunk);
                     count++;
                 }
+                // Flush remaining chunk
+                if (chunk.length > 0) {
+                    me.fire('data', chunk);
+                }
                 break;
             }
 
@@ -66,8 +77,13 @@ class Stream extends Base {
 
             for (const line of lines) {
                 if (line.trim()) {
-                    me.processLine(line);
+                    me.processLine(line, chunk);
                     count++;
+
+                    if (chunk.length >= chunkSize) {
+                        me.fire('data', chunk);
+                        chunk = [];
+                    }
                 }
             }
         }
@@ -77,11 +93,11 @@ class Stream extends Base {
 
     /**
      * @param {String} line
+     * @param {Array} chunk
      */
-    processLine(line) {
+    processLine(line, chunk) {
         try {
-            const data = JSON.parse(line);
-            this.fire('data', data);
+            chunk.push(JSON.parse(line));
         } catch (e) {
             console.warn('JSON parse error in Stream proxy:', e, line);
         }

test/playwright/unit/data/StoreProxy.spec.mjs

@@ -22,8 +22,7 @@ class MockProxy extends ProxyBase {
     }
 
     async read(operation) {
-        this.fire('data', {id: '1', name: 'Mock 1'});
-        this.fire('data', {id: '2', name: 'Mock 2'});
+        this.fire('data', [{id: '1', name: 'Mock 1'}, {id: '2', name: 'Mock 2'}]);
         return {success: true, count: 2};
     }
 }
@@ -48,7 +47,7 @@ test.describe.serial('Neo.data.Store Proxy Integration', () => {
         expect(store.proxy instanceof MockProxy).toBe(true);
     });
 
-    test('Store load() should use proxy', async () => {
+    test('Store load() should use proxy and progressive loading', async () => {
         const store = Neo.create(Store, {
             keyProperty: 'id',
             model: {
@@ -60,40 +59,20 @@ test.describe.serial('Neo.data.Store Proxy Integration', () => {
             }
         });
 
-        let loadFired = false;
-        store.on('load', () => loadFired = true);
+        let loadFiredCount = 0;
+        store.on('load', () => loadFiredCount++);
 
         await store.load();
 
-        expect(loadFired).toBe(true);
+        // Should fire load at least once during stream (progressive) and once at end?
+        // MockProxy fires data once (2 items).
+        // Store:
+        // 1. onData -> add -> isLoading=false -> fire('load') (Count: 1)
+        // 2. await proxy.read -> success -> fire('load') (Count: 2)
+        
+        expect(loadFiredCount).toBeGreaterThanOrEqual(1);
         expect(store.count).toBe(2);
         expect(store.get('1').name).toBe('Mock 1');
         expect(store.get('2').name).toBe('Mock 2');
     });
-
-    test('Store should suspend events during stream', async () => {
-        const store = Neo.create(Store, {
-            keyProperty: 'id',
-            proxy: {
-                module: MockProxy
-            }
-        });
-
-        let mutateCount = 0;
-        store.on('mutate', () => mutateCount++);
-
-        await store.load();
-
-        // mutate should only fire once (from add) or be suppressed if we used suspendEvents?
-        // logic: suspendEvents = true.
-        // proxy fires data -> onProxyData -> store.add(data) -> super.add -> splice -> fire 'mutate'
-        // if suspendEvents is true, Observable.fire checks !me.suspendEvents.
-        
-        // So mutate should NOT fire during streaming.
-        
-        // Wait, does store fire 'load' at the end? Yes.
-        // Does 'mutate' fire after suspendEvents = false? No, we don't queue events.
-        
-        expect(mutateCount).toBe(0);
-    });
 });

test/playwright/unit/data/proxy/Stream.spec.mjs

@@ -57,7 +57,8 @@ test.describe.serial('Neo.data.proxy.Stream', () => {
 
         const items = [];
         proxy.on('data', (data) => {
-            items.push(data);
+            // data is now an array
+            items.push(...data);
         });
 
         const result = await proxy.read({url: 'http://test.com/data.jsonl'});
@@ -98,7 +99,7 @@ ${JSON.stringify(mockData[1])}
 
         const items = [];
         proxy.on('data', (data) => {
-            items.push(data);
+            items.push(...data);
         });
 
         const result = await proxy.read({url: 'http://test.com/data.jsonl'});