Fix 67 - Removing processDataMemory()

action.yaml

@@ -37,10 +37,6 @@ inputs:
   property_path:
     description: 'Property path to be used by bucketizers'
     required: false
-  stream_data:
-    description: 'Boolean whether to stream the LDES members or the load them in memory'
-    required: false
-    default: 'false'
   timeout:
     description: 'Amount of time in milliseconds to wait for the datasource to fetch data in a single run'
     required: false
@@ -72,7 +68,6 @@ runs:
         INPUT_FRAGMENTATION_PAGE_SIZE: ${{ inputs.fragmentation_page_size }}
         INPUT_DATASOURCE_STRATEGY: ${{ inputs.datasource_strategy }}
         INPUT_PROPERTY_PATH: ${{ inputs.property_path }}
-        INPUT_STREAM_DATA: ${{ inputs.stream_data }}
         INPUT_TIMEOUT: ${{ inputs.timeout }}
         INPUT_ANNOUNCE: ${{ inputs.announce }}
     - name: Run post script

src/data-source-strategy/DatasourceContext.ts

@@ -1,6 +1,4 @@
 import type { EventStream } from '@treecg/actor-init-ldes-client';
-import type { Member, Bucketizer } from '@treecg/types';
-import type { Config } from '../utils/Config';
 import type Datasource from '../utils/interfaces/Datasource';
 
 class DatasourceContext {
@@ -14,10 +12,6 @@ class DatasourceContext {
     this.datasource = datasource;
   }
 
-  public getData(config: Config, bucketizer: Bucketizer): Promise<Member[]> {
-    return this.datasource.getData(config, bucketizer);
-  }
-
   public getLinkedDataEventStream(url: string, storage: string): EventStream {
     return this.datasource.getLinkedDataEventStream(url, storage);
   }

src/data-source-strategy/LDESClientDatasource.ts

@@ -1,49 +1,10 @@
 import type { EventStream } from '@treecg/actor-init-ldes-client';
 import { newEngine } from '@treecg/actor-init-ldes-client';
-import type { Member, Bucketizer } from '@treecg/types';
-import type { Config } from '../utils/Config';
 import type Datasource from '../utils/interfaces/Datasource';
 import type { LDESActionState } from '../utils/State';
-import { loadState, saveState } from '../utils/State';
+import { loadState } from '../utils/State';
 
 class LDESClientDatasource implements Datasource {
-  public async getData(
-    config: Config,
-    bucketizer: Bucketizer,
-  ): Promise<Member[]> {
-    return new Promise<Member[]>((resolve, reject) => {
-      try {
-        const ldes = this.getLinkedDataEventStream(config.url, config.storage);
-
-        const data: Member[] = [];
-
-        // If run takes longer than x minutes, pause the LDES Client
-        const timeout = setTimeout(() => ldes.pause(), config.timeout);
-
-        ldes.on('data', (member: Member) => {
-          bucketizer.bucketize(member.quads, member.id.value);
-          data.push(member);
-        });
-
-        ldes.on('now only syncing', () => {
-          timeout.unref();
-          console.log('now only syncing');
-          ldes.pause();
-        });
-
-        ldes.on('pause', () => {
-          saveState(ldes.exportState(), bucketizer.exportState(), config.storage);
-
-          console.log('No more data!');
-          resolve(data);
-        });
-      } catch (error: unknown) {
-        console.error(error);
-        return reject(error);
-      }
-    });
-  }
-
   public getLinkedDataEventStream(url: string, storage: string): EventStream {
     const options = {
       emitMemberOnce: true,

src/data.ts

@@ -1,7 +1,7 @@
 import { existsSync, mkdirSync } from 'fs';
 
 import type * as RDF from '@rdfjs/types';
-import type { Member, Bucketizer, RelationParameters } from '@treecg/types';
+import type { Member } from '@treecg/types';
 import DatasourceContext from './data-source-strategy/DatasourceContext';
 import LDESClientDatasource from './data-source-strategy/LDESClientDatasource';
 import BucketizerFragmentStrategy from './fragment-strategy/BucketizerFragmentStrategy';
@@ -15,7 +15,7 @@ export class Data {
   private readonly datasourceContext: DatasourceContext;
   private readonly fragmentContext: FragmentContext;
 
-  private RDFData: Member[];
+  private readonly RDFData: Member[];
 
   public constructor(config: Config) {
     this.config = config;
@@ -37,18 +37,7 @@ export class Data {
   }
 
   public processData(): Promise<void> {
-    return this.config.stream_data ?
-      this.processDataStreamingly() :
-      this.processDataMemory();
-  }
-
-  private async processDataMemory(): Promise<void> {
-    const bucketizer = await this.fragmentContext.getStrategy().initBucketizer(this.config);
-    await this.fetchData(bucketizer);
-
-    const hypermediaControls = bucketizer.getBucketHypermediaControlsMap();
-    const propertyPathQuads = bucketizer.getPropertyPathQuads();
-    await this.writeData(hypermediaControls, propertyPathQuads);
+    return this.processDataStreamingly();
   }
 
   private async processDataStreamingly(): Promise<void> {
@@ -105,56 +94,6 @@ export class Data {
     this.datasourceContext.setDatasource(new LDESClientDatasource());
   }
 
-  /**
-   * Fetch data using Datasource
-   */
-  public async fetchData(bucketizer: Bucketizer): Promise<void> {
-    return new Promise<void>(async (resolve, reject) => {
-      try {
-        this.RDFData = await this.datasourceContext.getData(
-          this.config,
-          bucketizer,
-        );
-        return resolve();
-      } catch (error: unknown) {
-        console.error(error);
-        return reject(error);
-      }
-    });
-  }
-
-  /**
-   * Write fetched data and hypermediacontrols to the output directory supplied in the config file
-   */
-  public writeData(
-    hypermediaControls: Map<string, RelationParameters[]>,
-    propertyPathQuads: RDF.Quad[],
-  ): Promise<void> {
-    return new Promise<void>(async (resolve, reject) => {
-      try {
-        const tasks: any[] = [];
-        this.RDFData.forEach(member => {
-          const extension = this.getOutputExtension(member.quads);
-          this.fragmentContext.setFileExtension(extension);
-
-          tasks.push(this.fragmentContext.fragment(member, this.config));
-        });
-
-        await Promise.all(tasks);
-        await this.fragmentContext.addHypermediaControls(
-          hypermediaControls,
-          propertyPathQuads,
-          this.config,
-        );
-
-        return resolve();
-      } catch (error: unknown) {
-        console.error(error);
-        return reject(error);
-      }
-    });
-  }
-
   private getOutputExtension(quads: RDF.Quad[]): FileExtension {
     const graphlessQuads = quads.filter(quad => quad.graph.termType === 'DefaultGraph');
     return graphlessQuads.length > 0 ? FileExtension.Turtle : FileExtension.TriG;

src/utils/Config.ts

@@ -31,9 +31,6 @@ export interface Config {
   // Property path that will be resolved by bucketizer
   property_path: string;
 
-  // Whether or not to process LDES member once received or to load all of them in memory
-  stream_data: boolean;
-
   // Amount of time to wait for the datasource to fetch data in a single run
   timeout: number;
 
@@ -56,7 +53,6 @@ export function getConfig(): Config {
     ),
     datasource_strategy: core.getInput('datasource_strategy'),
     property_path: core.getInput('property_path'),
-    stream_data: core.getBooleanInput('stream_data') || false,
     timeout: Number.parseInt(core.getInput('timeout'), 10) || 1_000 * 60 * 60 * 1,
     announce: core.getBooleanInput('announce') || false,
   };

src/utils/interfaces/Datasource.ts

@@ -1,9 +1,6 @@
 import type { EventStream } from '@treecg/actor-init-ldes-client';
-import type { Member, Bucketizer } from '@treecg/types';
-import type { Config } from '../Config';
 
 interface Datasource {
-  getData: (config: Config, bucketizer: Bucketizer) => Promise<Member[]>;
   getLinkedDataEventStream: (url: string, storage: string) => EventStream;
 }