Node-serialport: Sequential writing/reading

[JBtje]

💥 Proposal

Update documentation with below sample code

What feature you'd like to see

Basically #1679

Motivation

tldr; It took me way to long to figure out the answer to question #811

I've worked with serialport in C#, but had to port it to a Mac system, and thought it would be easiest to just use Electron and node-serialport. However, it took me way to long to figure out how it works, and how to make it work with sequential writing/reading, even though I think it is a very common use.

Write -> read -> wait for response -> write -> read -> wait for response.

Even though the wrapping code in stream.js uses Promises, it does not return promises, wich I think is too bad...
After many tries, and lots of babel errors, I went with just overwriting the methods, to make sure they DO return promises.

In short: It imports serialport the way you normally do, then it overwrites the essential methods (open / _write / _read) and makes sure they return a Promise.

./stream2.js

const debug  = require( 'debug' )( 'serialport/stream2' );

const SerialPort = require( 'serialport' );

function allocNewReadPool( poolSize ) {
    const pool = Buffer.allocUnsafe( poolSize );
    pool.used  = 0;
    return pool;
}

SerialPort.prototype.open = function( openCallback ) {
    if( this.isOpen ) {
        return this._asyncError( new Error( 'Port is already open' ), openCallback );
    }

    if( this.opening ) {
        return this._asyncError( new Error( 'Port is opening' ), openCallback );
    }

    this.opening = true;

    // Return a Promise, let the user chain ans handle exceptions
    return this.binding.open( this.path, this.settings ).then(
        () => {

            debug( 'opened', `path: ${this.path}` );
            this.opening = false;
            this.emit( 'open' );
            if( openCallback ) {
                openCallback.call( this, null );
            }
        },
        err => {
            this.opening = false;
            debug( 'Binding #open had an error', err );
            this._error( err, openCallback );
        },
    );
};

SerialPort.prototype._write = function( data, encoding, callback ) {
    if( !this.isOpen ) {
        console.error( 'Please open the connection first' );
        return false;
    }

    debug( '_write', `${data.length} bytes of data` );
    return this.binding.write( data ).then(
        () => {
            debug( 'binding.write', 'write finished' );
        },
        err => {
            debug( 'binding.write', 'error', err );
            if( !err.canceled ) {
                this._disconnected( err );
            }
        },
    );
};

SerialPort.prototype._read = function( bytesToRead ) {
    debug( 'isOpen: ', this.isOpen );
    if( !this.isOpen ) {
        console.error( 'Please open the connection first' );
        return false;
    }

    if( !this._pool || this._pool.length - this._pool.used < this._kMinPoolSpace ) {
        debug( '_read', 'discarding the read buffer pool' );
        this._pool = allocNewReadPool( this.settings.highWaterMark );
    }

    // Grab another reference to the pool in the case that while we're
    // in the thread pool another read() finishes up the pool, and
    // allocates a new one.
    const pool   = this._pool;
    // Read the smaller of rest of the pool or however many bytes we want
    const toRead = Math.min( pool.length - pool.used, bytesToRead );
    const start  = pool.used;

    // the actual read.
    debug( '_read', `reading` );
    return this.binding.read( pool, start, toRead ).then(
        bytesRead => {
            debug( 'binding.read', `finished` );
            // zero bytes means read means we've hit EOF? Maybe this should be an error
            if( bytesRead === 0 ) {
                debug( 'binding.read', 'Zero bytes read closing readable stream' );
                this.push( null );
                return;
            }

            pool.used += bytesRead;
            this.push( pool.slice( start, start + bytesRead ) );
        },
        err => {
            debug( 'binding.read', `error`, err );
            if( !err.canceled ) {
                this._disconnected( err );
            }
            this._read( bytesToRead ); // prime to read more once we're reconnected
        },
    );
};

SerialPort.list = function( callback ) {
    if( !SerialPort.Binding ) {
        throw new TypeError( 'No Binding set on `SerialPort.Binding`' );
    }

    const promise = SerialPort.Binding.list();
    if( typeof callback === 'function' ) {
        promise.then( ports => callback( null, ports ), err => callback( err ) );
    }
    return promise;
};


module.exports = SerialPort;

main.js

const SerialPort = require( './stream2' );

let comName = 'COM1';

this.SerialPort = new SerialPort( comName, {
    autoOpen:      false,
    baudRate:      115200,
    dataBits:      8,
    highWaterMark: 64,
    stopBits:      1,
    parity:        'none',
} );

this.SerialPort.open()
    .then( () => {
            this.SerialPort._write( Buffer.from( 'AA000001', 'hex' ) )
                .then(
                    async r => {

                        // Read the header, which contains the length of the package
                        await this.SerialPort._read( 2 );
                        console.log( 'Header: ', this.SerialPort._pool );

                        let L = this.SerialPort._pool[1] - 1;
                        // Read the package
                        await this.SerialPort._read( L );

                        // Note: data is stored in "this.SerialPort._pool", but only until you do another read... So
                        // make sure to get it out of there
                        let Package = this.SerialPort._pool.slice( 0, L );
                        console.log( 'Package: ', Package );
                    },
                )
                .catch(
                    err => {
                        console.error( 'Error 1', err );
                        console.error( 'Closing the connection' );
                        this.SerialPort.close( null, null );
                    },
                )
            ;
        },
    )
    .then( () => {
            this.SerialPort._write( Buffer.from( 'AA000002', 'hex' ) )
                .then(
                    async r => {

                        // Read the header, which contains the length of the package
                        await this.SerialPort._read( 2 );
                        console.log( 'Header: ', this.SerialPort._pool );

                        let L = this.SerialPort._pool[1] - 1;
                        // Read the package
                        await this.SerialPort._read( L );

                        // Note: data is stored in "this.SerialPort._pool", but only until you do another read... So
                        // make sure to get it out of there
                        let Package = this.SerialPort._pool.slice( 0, L );
                        console.log( 'Package: ', Package );
                    },
                )
                .catch(
                    err => {
                        console.error( 'Error 1', err );
                        console.error( 'Closing the connection' );
                        this.SerialPort.close( null, null );
                    },
                )
            ;
        },
    )
    .catch( ( error ) => {
        console.error( 'Any Error?', error );
    } )
    .then(
        r => {
            // Finally, close the connection
            console.log( 'Closing the PORT' );
            this.SerialPort.close( null, null );
        },
    )
    .catch( ( error ) => {
        console.error( 'Final Error', error );
        console.error( 'Closing the connection' );
        this.SerialPort.close( null, null );
    } );

_No guarantee it works, it is a cleaned up code dump (not tested afterwards). I hope I can help someone with it._

[HipsterBrown]

@JBtje To help clean up this code, you can also use the Serialport bindings as a separate package and avoid streams altogether. https://www.npmjs.com/package/@serialport/bindings

You can see the list of methods available to the bindings via the AbtractBindings class documentation -> https://serialport.io/docs/en/api-binding-abstract

This will also eliminate the internal pool, so you can set the value returned by bindings.read to a variable instead of reading from the internal pool after calling .read.

[reconbot]

Yeah the bindings is the way to use promises right now, it's pretty full featured. they're the lower level underneath the @serialport/stream package. If we changed the api of the stream it wouldn't be a nodejs stream anymore.

If you need to work with streams I do recommend this BlueStream method https://www.npmjs.com/package/bluestream#readasync for reading some information off of a stream.

[JBtje]

Thank you both for the reply. Before I ended with the above (hack), I examined your code and thought that the @serialport/bindings was the way to go. I started with a very simple (i thought) method: copy serialport/lib/index.js and serialport/lib/parser.js and move it to my own folder so I could modify it.

But "moving" the files was enough reason for Babel to complain with exceptions about not being able to import the module, and many hours later I gave up :(

So, for my understanding: I can make a simple wrapper around @serialport/bindings, which implements the methods of https://serialport.io/docs/en/api-binding-abstract ?

Regarding "so you can set the value returned by bindings.read to a variable", could you give me a hint?
My assumption is that, if I do the above, I'll end up with something like this

function read( bytesToRead ) {
    ...
    return this.binding.read( pool, start, toRead ).then(
        bytesRead => {
            if( bytesRead === 0 ) {
                return;
            }

            this.package = pool.slice( start, start + bytesRead ) );
        },
        err => {
            ...
        },
    );
}

My problem with this is, that bytesRead appears out of nowhere. I assume that this.bindings.read() calls the read method in e.g. @serialport/bindings/lib/win32.js:

  read(buffer, offset, length) {
    return super
      .read(buffer, offset, length)
      .then(() => promisify(binding.read)(this.fd, buffer, offset, length))
      .catch(err => {
        if (!this.isOpen) {
          err.canceled = true
        }
        throw err
      })
  }

but its a bit magical to me where bytesRead comes from.
Anyhow, how would I rewrite the return this.binding.read( pool, start, toRead ).then( code so that it returns this.package.

I now see I still have this.pool in the code; is it even possible to remove that? Bindings.read still needs a Buffer right?

[HipsterBrown]

Here is an example of using the @serialport/bindings package by itself:

const Binding = require('@serialport/bindings');
const comName = 'COM1';
const openOptions = {
    baudRate:      115200,
    dataBits:      8,
    stopBits:      1,
    parity:        'none',
};

const binding = new Binding();

// allows using await
(async () => {
  await binding.open(comName, openOptions);
  await binding.write(Buffer.from( 'AA000002', 'hex' ));

  // buffer to read header info into
  const header = Buffer.alloc(2);
  const bytesRead = await binding.read(header, 0, 2);
  console.log( 'Header: ', header );

  const L = header[1] - 1;
  // buffer to read package data into
  const package = Buffer.alloc(L);
  await binding.read(package, 0, L);
  console.log( 'Package: ', package );

  await binding.close();
})();

The read method takes 3 arguments:

• buffer, the buffer for the data to be written to
• offset, offset in the buffer to start writing at
• length, an integer specifying the number of bytes to read

It returns the number of bytesRead into the buffer. (similar to https://nodejs.org/dist/latest-v10.x/docs/api/fs.html#fs_fs_read_fd_buffer_offset_length_position_callback)

I hope this helps clear up how to use the @serialport/bindings package for your use case.

[JBtje]

in one word: amazing

its even less code then I expected, and I now understand the bytesRead part: thank you!

I would vote to add that code to the docs!

[HipsterBrown]

Happy to help!

I would vote to add that code to the docs!

I'll add it to my list for this weekend. 😄

[JBtje]

Perhaps nice to add with the above doc, is that the user needs to run:
npm install @serialport/bindings

If not, you'll see the exception:
Uncaught NodeError: The "path" argument must be of type string. Received type undefined at assertPath (path.js:39:11) at dirname (path.js:651:5)

And when using Electron, chance is that you'll need to run ./node_modules/.bin/electron-rebuild as well.

Note:
const binding = new Binding();
needs to be
const binding = new Binding( openOptions );

Edit
Let's add some code to list the devices as well

Binding.list().then(
    ports => ports.forEach( console.log ),
    err => console.error( err ),
);

[HipsterBrown]

@JBtje Good notes.

For Electron, we have instructions about electron-rebuild in the installation docs: https://serialport.io/docs/en/guide-installation#electron