Add scripted-transport harness and Modbus golden-frame tests

.github/workflows/ci.yml

@@ -48,6 +48,9 @@ jobs:
     - working-directory: node-drivers
       run: node test/integration/eip.js
 
+    - working-directory: node-drivers
+      run: node test/integration/cip-connected.js
+
     - uses: actions/setup-python@v5
       with:
         python-version: '3.11'

src/core/modbus/pdu.js

@@ -46,7 +46,9 @@ export default class PDU {
         if (Buffer.isBuffer(value) && value.length === 2) {
           value.copy(buffer, offset, 0, 2);
         } else if (Number.isFinite(value)) {
-          buffer.writeInt16BE(value, offset);
+          /** registers are unsigned 16-bit on the wire; mask so negative
+           * inputs encode as two's complement instead of throwing */
+          buffer.writeUInt16BE(value & 0xFFFF, offset);
         } else {
           throw new Error('Modbus write request error: currently supports buffer, array of 2-byte buffers, or array of finite numbers');
         }
@@ -58,6 +60,25 @@ export default class PDU {
     return buffer;
   }
 
+  /**
+   * Function 0x0F layout: fn(1), address(2), quantity of outputs(2),
+   * byte count(1), coil values packed LSB-first
+   */
+  static EncodeWriteMultipleCoilsRequest(address, values) {
+    const byteCount = Math.ceil(values.length / 8);
+    const buffer = Buffer.alloc(6 + byteCount);
+    buffer.writeUInt8(Functions.WriteMultipleCoils, 0);
+    buffer.writeUInt16BE(address, 1);
+    buffer.writeUInt16BE(values.length, 3);
+    buffer.writeUInt8(byteCount, 5);
+    for (let i = 0; i < values.length; i++) {
+      if (values[i]) {
+        buffer[6 + (i >> 3)] |= 1 << (i & 0b111);
+      }
+    }
+    return buffer;
+  }
+
   static Decode(buffer, offsetRef, pduLength) {
     const fn = PDU.Fn(buffer, offsetRef);
     const data = PDU.Data(buffer, offsetRef, pduLength);

src/defragger.js

@@ -8,22 +8,27 @@ export default class Defragger {
     this._lengthHandler = lengthHandler;
   }
 
+  /**
+   * Appends data, if given, and returns the next complete frame or null.
+   * Call again without data to drain any remaining buffered frames.
+   */
   defrag(data) {
-    let defraggedData = null;
-
-    this._dataLength += data.length;
-    this._data = Buffer.concat([this._data, data], this._dataLength);
+    if (data != null && data.length > 0) {
+      this._dataLength += data.length;
+      this._data = Buffer.concat([this._data, data], this._dataLength);
+    }
 
-    while (
+    if (
       this._dataLength > 0
       && this._completeHandler(this._data, { current: 0 }, this._dataLength)
     ) {
       const length = this._lengthHandler(this._data, { current: 0 });
-      defraggedData = this._data.slice(0, length);
+      const frame = this._data.slice(0, length);
       this._dataLength -= length;
       this._data = this._data.slice(length);
+      return frame;
     }
 
-    return defraggedData;
+    return null;
   }
 }

src/layers/Layer.js

@@ -145,8 +145,14 @@ export default class Layer extends EventEmitter {
 
   static forwardTo(layer, data, info, context) {
     if (layer._defragger != null) { // eslint-disable-line no-underscore-dangle
-      data = layer._defragger.defrag(data); // eslint-disable-line no-underscore-dangle
-      if (data == null) return;
+      /** one chunk may complete several frames; forward each one */
+      let frame = layer._defragger.defrag(data); // eslint-disable-line no-underscore-dangle
+      while (frame != null) {
+        layer.emit('data', frame, info, context);
+        layer.handleData(frame, info, context);
+        frame = layer._defragger.defrag(); // eslint-disable-line no-underscore-dangle
+      }
+      return;
     }
     layer.emit('data', data, info, context);
     layer.handleData(data, info, context);

src/layers/cip/layers/EIP/cpf.js

@@ -126,7 +126,7 @@ class Packet {
           value.flags.supportsCIPClass0or1UDPBasedConnections = !!getBits(flags, 8, 9);
 
           let nameLength;
-          for (nameLength = 0; nameLength <= 16; nameLength++) {
+          for (nameLength = 0; nameLength < 16; nameLength++) {
             if (buffer[offsetRef.current + nameLength] === 0) {
               break;
             }

src/layers/cip/layers/EIP/index.js

@@ -307,8 +307,8 @@ export default class EIPLayer extends Layer {
           clearTimeout(timeoutHandler);
           if (hostsSpecified) {
             timeoutHandler = setTimeout(finalizer, resetTimeout);
-            return;
-            // return true;
+            /** keep this callback registered for replies from the remaining hosts */
+            return true;
           }
           finalizer();
         } else {

src/layers/cip/layers/EIP/packet.js

@@ -158,7 +158,7 @@ export default class EIPPacket {
     buffer.writeUInt16LE(command, OFFSET_COMMAND);
     buffer.writeUInt16LE(dataLength, OFFSET_DATA_LENGTH);
     buffer.writeUInt32LE(sessionHandle, OFFSET_SESSION_HANDLE);
-    buffer.writeUInt32LE(status.code, OFFSET_STATUS);
+    buffer.writeUInt32LE(status, OFFSET_STATUS);
     (senderContext || NullSenderContext).copy(buffer, OFFSET_SENDER_CONTEXT, 0, 8);
     buffer.writeUInt32LE(options, OFFSET_OPTIONS);
     if (dataLength > 0) {

src/layers/cip/layers/Logix5000/index.js

@@ -53,14 +53,14 @@ import {
 
 const DEFAULT_SCOPE = '__DEFAULT_GLOBAL_SCOPE__';
 
-function Logix5000DecodeDataType(buffer, offsetRef, cb) {
+function Logix5000DecodeDataType(buffer, offsetRef) {
   const startingOffset = offsetRef.current;
-  const type = EPath.Decode(buffer, offsetRef, null, false, cb);
+  const segment = EPath.Decode(buffer, offsetRef, null, false);
   /** TODO: Why is this necessary? */
   if (offsetRef.current - startingOffset < 2) {
     offsetRef.current += 1;
   }
-  return type;
+  return segment;
 }
 
 async function readTagFragmented(layer, path, elements) {
@@ -69,20 +69,21 @@ async function readTagFragmented(layer, path, elements) {
   const reqData = Buffer.allocUnsafe(6);
   reqData.writeUInt16LE(elements, 0);
 
-  const offsetRef = { current: 0 };
+  let requestOffset = 0;
   const chunks = [];
 
   while (true) {
-    reqData.writeUInt32LE(offsetRef.current, 2);
+    reqData.writeUInt32LE(requestOffset, 2);
     const reply = await sendPromise(layer, service, path, reqData, 5000);
 
-    /** remove the tag type bytes if already received */
+    /** each reply starts with the tag type; keep it on the first chunk only */
+    const offsetRef = { current: 0 };
     Logix5000DecodeDataType(reply.data, offsetRef);
     const dataTypeOffset = offsetRef.current;
     chunks.push(chunks.length > 0 ? reply.data.slice(dataTypeOffset) : reply.data);
 
     if (reply.status.code === GeneralStatusCodes.PartialTransfer) {
-      offsetRef.current = reply.data.length - dataTypeOffset;
+      requestOffset += reply.data.length - dataTypeOffset;
     } else if (reply.status.code === 0) {
       break;
     } else {
@@ -101,7 +102,7 @@ async function parseReadTagMemberStructure(layer, structureType, data, offset) {
 
   const template = await layer.readTemplate(structureType.template.id);
   if (!template || !Array.isArray(template.members)) {
-    return new Error(`Unable to read template: ${structureType.template.id}`);
+    throw new Error(`Unable to read template: ${structureType.template.id}`);
   }
 
   const { members } = template;
@@ -153,15 +154,15 @@ async function parseReadTag(layer, scope, tag, elements, data) {
     return undefined;
   }
 
-  let typeInfo;
-  const offset = Logix5000DecodeDataType(data, 0, (val) => { typeInfo = val.value; });
+  const offsetRef = { current: 0 };
+  const typeSegment = Logix5000DecodeDataType(data, offsetRef);
+  const typeInfo = typeSegment ? typeSegment.value : undefined;
 
   if (!typeInfo) {
     throw new Error('Unable to decode data type from read tag response data');
   }
 
   const values = [];
-  const offsetRef = { current: offset };
 
   if (!typeInfo.constructed || typeInfo.abbreviated === false) {
     for (let i = 0; i < elements; i++) {
@@ -225,23 +226,27 @@ async function parseReadTag(layer, scope, tag, elements, data) {
 
 function statusHandler(code, extended, cb) {
   let error = GenericServiceStatusDescriptions[code];
-  if (typeof error === 'object' && Buffer.isBuffer(extended) && extended.length >= 0) {
-    error = error[extended.readUInt16LE(0)];
+  if (typeof error === 'object') {
+    if (Buffer.isBuffer(extended) && extended.length >= 2) {
+      error = error[extended.readUInt16LE(0)];
+    } else {
+      error = undefined;
+    }
   }
   if (error) {
     cb(null, error);
   }
 }
 
 /** Use driver specific error handling if exists */
-async function send(self, service, path, data, callback /* , timeout */) {
+async function send(self, service, path, data, callback, timeout) {
   try {
     const request = new CIPRequest(service, path, data, null, {
       serviceNames: SymbolServiceNames,
       statusHandler,
     });
 
-    const response = await self.sendRequest(true, request);
+    const response = await self.sendRequest(true, request, null, timeout);
     // console.log(response);
     if (response.status.error) {
       callback(response.status.description, response);
@@ -395,11 +400,11 @@ function parseTemplateNameInfo(data, offset, cb) {
 //   return error;
 // }
 
-function scopedGenerator() {
+function scopedGenerator(...scopeArgs) {
   const separator = '::';
-  const args = [...arguments].filter((arg) => !!arg);
-  const preface = args.length > 0 ? args.join(separator) + separator : '';
-  return () => preface + [...arguments].join(separator);
+  const scopes = scopeArgs.filter((arg) => !!arg);
+  const preface = scopes.length > 0 ? scopes.join(separator) + separator : '';
+  return (...parts) => preface + parts.join(separator);
 }
 
 async function getSymbolInstanceID(layer, scope, tag) {
@@ -886,7 +891,7 @@ export default class Logix5000 extends CIPLayer {
       }
 
       for (let i = 0; i < sizeOfMasks; i++) {
-        if (ORmasks[i] < 0 || ORmasks > 0xFF || ANDmasks[i] < 0 || ANDmasks > 0xFF) {
+        if (ORmasks[i] < 0 || ORmasks[i] > 0xFF || ANDmasks[i] < 0 || ANDmasks[i] > 0xFF) {
           resolver.reject('Values in masks must be greater than or equal to zero and less than or equal to 255');
           return;
         }

src/layers/cip/layers/internal/CIPConnectionLayer.js

@@ -488,6 +488,7 @@ class CIPConnectionLayer extends Layer {
   }
 
   handleDestroy() {
+    stopResend(this);
     this._connectionState = 0;
     this._sequenceCount = 0;
     this.sendInfo = null;

src/layers/cip/layers/internal/CIPInternalLayer.js

@@ -27,10 +27,8 @@ class CIPInternalLayer extends Layer {
     }
   }
 
-  sendRequest(connected, request, callback) {
+  sendRequest(connected, request, callback, timeout) {
     return CallbackPromise(callback, (resolver) => {
-      const timeout = null;
-
       const context = this.contextCallback((error, message) => {
         try {
           if (error) {

src/layers/modbus/index.js

@@ -10,7 +10,6 @@ const {
   ReadInputRegisters,
   ReadHoldingRegisters,
   WriteSingleCoil,
-  WriteMultipleCoils,
   WriteSingleHoldingRegister,
   // WriteMultipleHoldingRegisters
 } = MB.Functions;
@@ -22,6 +21,13 @@ const DefaultOptions = {
 };
 
 function readRequest(self, fn, address, count, callback) {
+  if (typeof count === 'function' && callback == null) {
+    callback = count; // eslint-disable-line no-param-reassign
+    count = undefined; // eslint-disable-line no-param-reassign
+  }
+  if (count == null) {
+    count = 1; // eslint-disable-line no-param-reassign
+  }
   return CallbackPromise(callback, (resolver) => {
     self._send(PDU.EncodeReadRequest(fn, address, count), {}, resolver);
   });
@@ -92,20 +98,20 @@ export default class Modbus extends Layer {
     return readRequest(this, ReadInputRegisters, inputAddressing, count, callback);
   }
 
-  readHoldingRegisters(inputAddressing, count = 1, callback) {
+  readHoldingRegisters(inputAddressing, count, callback) {
     return readRequest(this, ReadHoldingRegisters, inputAddressing, count, callback);
   }
 
   writeSingleCoil(inputAddressing, value, callback) {
-    const values = [value ? 0x00FF : 0x0000];
+    /** 0xFF00 is the only valid ON value for function 0x05 */
+    const values = [value ? 0xFF00 : 0x0000];
     return writeRequest(this, WriteSingleCoil, inputAddressing, values, callback);
   }
 
   writeMultipleCoils(inputAddressing, values, callback) {
-    for (let i = 0; i < values.length; i++) {
-      values[i] = values[i] ? 0x00FF : 0x0000;
-    }
-    return writeRequest(this, WriteMultipleCoils, inputAddressing, values, callback);
+    return CallbackPromise(callback, (resolver) => {
+      this._send(PDU.EncodeWriteMultipleCoilsRequest(inputAddressing, values), {}, resolver);
+    });
   }
 
   writeSingleHoldingRegister(inputAddressing, values, callback) {

src/layers/pccc/encoding.js

@@ -65,7 +65,7 @@ export function EncodeDataDescriptor(data, offsetRef, id, size) {
   }
 
   if (idLength > 0 && sizeLength === 0) {
-    offsetRef.current = data.writeUInt8(((0b1000 | idLength) << 4) | size, offsetRef);
+    offsetRef.current = data.writeUInt8(((0b1000 | idLength) << 4) | size, offsetRef.current);
     offsetRef.current = encodeUnsignedInteger(data, offsetRef.current, id, idLength);
     return;
   }

src/layers/pccc/index.js

@@ -248,8 +248,10 @@ export default class PCCCLayer extends Layer {
       const callback = this.callbackForContext(savedContext.context);
       if (callback != null) {
         callback(getError(packet.status), packet, info);
-        return;
       }
+      /** internal replies must never be forwarded to upper layers,
+       * even if the callback was already consumed */
+      return;
     }
 
     this.forward(packet.data, info, savedContext.context);

src/layers/pccc/packet.js

@@ -229,6 +229,10 @@ export default class PCCCPacket {
       throw new Error(`Unsupported address: ${address}`);
     }
 
+    if (info.id == null) {
+      throw new Error(`Writing to ${info.datatype} files is not currently supported (address: ${address})`);
+    }
+
     const valueCount = values.length;
     const dataValueLength = valueCount * info.size;
     const dataTypeLength = DataTypeEncodingLength(info.id, info.size);

src/layers/tcp/index.js

@@ -240,8 +240,11 @@ export default class TCPLayer extends Layer {
         });
       }
     } else if (this._connectionState === 0) {
-      /** Reconnect */
-      connect(this);
+      /** Reconnect only when there is something to send, otherwise a
+       * deferred wakeup after close() would reopen the connection */
+      if (this.hasRequest()) {
+        connect(this);
+      }
     }
   }