Merge branch 'RTDE-Driver' of https://github.com/Harrier-SA/amrc-connectivity-stack into RTDE-Driver

Author: Harrier-SA

acs-edge/lib/devices/MTConnect.ts

@@ -18,6 +18,8 @@ export class MTConnectConnection extends (
     #connDetails: restConnDetails
     #cancelToken: CancelTokenStatic;
     #source: CancelTokenSource
+    #nextSequence: number | null = null;
+    #pollInterval: NodeJS.Timeout | null = null;
 
     constructor(type: string, connDetails: restConnDetails) {
         super(type, connDetails);
@@ -38,66 +40,79 @@ export class MTConnectConnection extends (
     }
 
     /**
-     * Opens http connection stream and emits data response
-     * @param pollInt Connection Polling interval
+     * Polls MTConnect endpoints at regular intervals
+     * @param metrics Metrics object containing endpoint addresses
+     * @param interval Polling interval in milliseconds
      */
     async sample(metrics: Metrics, interval: number) {
-        let res = await this.get(
-            `/sample?interval=0&heartbeat=1000`,
-            {
-                responseType: "stream",
-                cancelToken: this.#source.token
-            }
-        );
-        let headersCompleted = false;
-        let bodyCompleted = false;
-        let buffer = Buffer.alloc(0);
-        let receivedBodyChunk = 0;
-        let contentLength = 0;
-        const pass = new PassThrough();
-        pass.on('data', (chunk) => {
-            if (!headersCompleted) {
-                const headers = chunk.toString().split(/\r?\n/);
-                if (headers.length >= 3) {
-                    try {
-                        contentLength = parseInt(headers[2].split(":")[1]);
-                        if (!isNaN(contentLength)) {
-                            buffer = Buffer.alloc(contentLength);
-                            headersCompleted = true;
-                        }
-                    } catch (e) {
-                        console.log(e);
-                    }
+        // Get all unique addresses
+        const endpoints = [...new Set(metrics.addresses)];
+        
+        log(`[MTConnect] Starting polling of ${endpoints.length} endpoint(s) every ${interval}ms`);
+        
+        // Function to fetch data from a specific endpoint
+        const pollEndpoint = async (endpoint: string) => {
+            try {
+                // Build URL with sequence parameter if available
+                let url = endpoint;
+                
+                // Add sequence parameter for /sample endpoints
+                if (endpoint.includes('/sample') && this.#nextSequence) {
+                    url = endpoint.includes('?')
+                        ? `${endpoint}&from=${this.#nextSequence}`
+                        : `${endpoint}?from=${this.#nextSequence}`;
                 }
-            }
-            if (receivedBodyChunk < contentLength || isNaN(contentLength)) {
-                const contentStartOffset = Math.max(chunk.indexOf("<?xml", 0, 'utf8'), 0);
-                try {
-                    chunk.copy(buffer, receivedBodyChunk, contentStartOffset, chunk.byteLength);
-                } catch (err) {
-                    console.log(err);
+                
+                log(`[MTConnect] Polling: ${url}`);
+                
+                let res = await this.get(url, {
+                    cancelToken: this.#source.token
+                });
+
+                // Convert response to string
+                const xmlData = typeof res.data === 'string' 
+                    ? res.data 
+                    : res.data.toString();
+
+                // Extract and update nextSequence for future requests
+                const nextSeqMatch = xmlData.match(/nextSequence="(\d+)"/);
+                if (nextSeqMatch) {
+                    this.#nextSequence = parseInt(nextSeqMatch[1]);
+                    log(`[MTConnect] Updated sequence to: ${this.#nextSequence}`);
                 }
-                receivedBodyChunk += chunk.byteLength - contentStartOffset;
 
-                if (receivedBodyChunk === contentLength) {
-                    bodyCompleted = true;
+                // Emit data only for this specific endpoint
+                let obj: any = {};
+                obj[endpoint] = Buffer.from(xmlData);
+                this.emit('data', obj);
+
+            } catch (err: any) {
+                // Log detailed error information including the problematic address
+                if (err.response) {
+                    // HTTP error (4xx, 5xx)
+                    log(`[MTConnect] ERROR - Invalid address or HTTP error for endpoint: ${endpoint}`);
+                    log(`[MTConnect] HTTP Status: ${err.response.status} ${err.response.statusText}`);
+                    log(`[MTConnect] Response: ${JSON.stringify(err.response.data)}`);
+                } else if (err.request) {
+                    // Request was made but no response received (network error, timeout, etc.)
+                    log(`[MTConnect] ERROR - No response from endpoint: ${endpoint}`);
+                    log(`[MTConnect] Possible causes: incorrect hostname/IP, network issue, or service not running`);
+                    log(`[MTConnect] Details: ${err.message}`);
+                } else {
+                    // Something else went wrong
+                    log(`[MTConnect] ERROR - Failed to poll endpoint: ${endpoint}`);
+                    log(`[MTConnect] Error: ${err.message}`);
                 }
             }
-            if (bodyCompleted) {
-                // this.emit("asyncData", buffer);
-                let payload = Buffer.from(buffer);
-                metrics.addresses.forEach(addr => {
-                    let obj: any = {};
-                    obj[addr] = payload;
-                    this.emit('data', obj);
-                })
-                headersCompleted = false;
-                bodyCompleted = false;
-                buffer = Buffer.alloc(0);
-                receivedBodyChunk = 0;
-            }
-        });
-        res.data.pipe(pass);
+        };
+
+        // Poll all endpoints immediately on start
+        await Promise.all(endpoints.map(endpoint => pollEndpoint(endpoint)));
+
+        // Then poll all endpoints at regular intervals
+        this.#pollInterval = setInterval(async () => {
+            await Promise.all(endpoints.map(endpoint => pollEndpoint(endpoint)));
+        }, interval);
     }
 
     open() {
@@ -107,6 +122,13 @@ export class MTConnectConnection extends (
 
     close(): void {
         log("Closing MTConnectConnection");
+        
+        // Clear the polling interval
+        if (this.#pollInterval) {
+            clearInterval(this.#pollInterval);
+            this.#pollInterval = null;
+        }
+        
         this.#source.cancel("Stream Close Requested");
         this.emit("close");
     }