Multipart http encoder - feature request design review

[Bart Butenaers]

Hi folks,

I started yesterday experimenting with the httpresponse (= httpout) node, to adapt it for creating multipart http streams.  

The code is not complete yet !   But it seems to be working fine for my camera's...

I will explain my prove of concept in detail, and hopefully it is understandable.

Multipart stream decoder

A couple of months ago, I created a pull request for multipart http streaming: the httprequest node was modified to be able to decode an endless http response stream.  E.g. specify an MJPEG url of your camera, and the httprequest node will return an endless stream of images on it's output port.  Last week Nick gave me some homework: adding some unit tests for this pull request.  To be able to unit test this stream decoder, I needed to have a test stream (i.e. an endless stream of images).  If I'm not mistaken, these are the only options I have:

• Have the unit test code point to a public url (e.g. a public webcam).  However when that camera is down, the unit test would start failing.
• Add some encoder code to the unit test, to create a local mjpeg stream.  However it seems a bit silly, since I would like to have to have an encoder in Node-Red (see next option).
• Add an encoder to Node-Red (httpresponse node), so it can be used by everybody.  And afterwards I can use it also in the unit test of the mjpeg decoder.

Multipart stream encoder

Purpose is to create a multipart encoder, e.g. to create an mjpeg stream.  So when Node-red gets a particular http request from the outside world, we want to return a stream (i.e. an infinite response).

Currently the httpin/httpresponse nodes are behaving like this (without my modification):

1. An url is entered, pointing to our Node-Red installation
2. An http request arrives in ExpressJs 
3. Since it maps to the URL path in our httpin node, the request is passed to that node
4. The httpin node will add the (wrapped) request object in the msg.payload field, and add the response object in the msg.res fields
5. In the template node e.g. some html can be provided
6. The httpresponse node sends the content to the response object
7. The http response is returned to the browser where it will be rendered for the user

Thus a single http request results in a single (finite) http response.  

The above setup in Node-Red is very simple and powerful.  Brilliant!  That is why I love Node-Red ...

So I wanted to reuse this setup, and keep it as simple as possible.  

This is what happens when you select the new 'streaming' checkbox in the httpresponse node:

1. An (stream) url is entered, pointing to our Node-Red installation
2. An http request arrives in ExpressJs 
3. Since it maps to the URL path in our httpin node, it is passed to that node
4. The httpin node will add the (wrapped) request object in the msg.payload field, and add the response object in the msg.res fields
5. The httpresponse node is going to store the response (wrapper) - from msg.res - in an array for using it later on.  Moreover a disconnection handler is added to the response, which will remove the response from our array as soon as the client (e.g. browser) disconnects.
6. Another node sends images to the httpresponse node (those messages don't contain a msg.res field).
7. The http response node will send the images to ALL the response objects that are available in the array
8. Since the http response is not ended by the httpresponse node, the response has become endless (<headers> <image> <boundary> <headers> <image> <boundary> <headers> <image> <boundary> ...)
9. The mjpeg camera stream is returned to the browser, where it will be rendered to the user.

So a single http request results now in an endless http response, until the client closes the connection.

This is the code for my prove of concept (in the 21-httpin.html file only a single checkbox is added) in the 21-httpin.js file (in the commented sections):

    function HTTPOut(n) {
        RED.nodes.createNode(this,n);
        var node = this;
        this.headers = n.headers||{};
        this.statusCode = n.statusCode;
        // ********************************** START CHANGE ***********************************
        this.streaming = n.streaming || false;
        this.responses = [];
        const boundary = 'myboundary';
        // ********************************** END CHANGE ***********************************      
       
        this.on("input",function(msg) {
            if (msg.res) {
                // ********************************** START CHANGE ***********************************
                if (node.streaming == true) {
                    node.responses.push(msg.res);
                    // Handle a disconnect (i.e. remove the response object from the array)
                    msg.res._res.connection.on('close', function() {
                        var index = node.responses.indexOf(msg.res);

                        if (index > -1) {
                            node.responses.splice(index, 1);
                        }
                    });
                     
                    msg.res._res.writeHead(200, {
                        'Content-Type': 'multipart/x-mixed-replace;boundary=--' + boundary,
                        'Connection': 'keep-alive',
                        'Expires': 'Fri, 01 Jan 1990 00:00:00 GMT',
                        'Cache-Control': 'no-cache, no-store, max-age=0, must-revalidate',
                        'Pragma': 'no-cache'
                    })
                }        
                // ********************************** END CHANGE ***********************************                
               
                var headers = RED.util.cloneMessage(node.headers);
                if (msg.headers) {
                    if (msg.headers.hasOwnProperty('x-node-red-request-node')) {
                        var headerHash = msg.headers['x-node-red-request-node'];
                        delete msg.headers['x-node-red-request-node'];
                        var hash = hashSum(msg.headers);
                        if (hash === headerHash) {
                            delete msg.headers;
                        }
                    }
                    if (msg.headers) {
                        for (var h in msg.headers) {
                            if (msg.headers.hasOwnProperty(h) && !headers.hasOwnProperty(h)) {
                                headers[h] = msg.headers[h];
                            }
                        }
                    }
                }
                if (Object.keys(headers).length > 0) {
                    msg.res._res.set(headers);
                }
                if (msg.cookies) {
                    for (var name in msg.cookies) {
                        if (msg.cookies.hasOwnProperty(name)) {
                            if (msg.cookies[name] === null || msg.cookies[name].value === null) {
                                if (msg.cookies[name]!==null) {
                                    msg.res._res.clearCookie(name,msg.cookies[name]);
                                } else {
                                    msg.res._res.clearCookie(name);
                                }
                            } else if (typeof msg.cookies[name] === 'object') {
                                msg.res._res.cookie(name,msg.cookies[name].value,msg.cookies[name]);
                            } else {
                                msg.res._res.cookie(name,msg.cookies[name]);
                            }
                        }
                    }
                }
               
                if (node.streaming == false) {
                    var statusCode = node.statusCode || msg.statusCode || 200;
                    if (typeof msg.payload == "object" && !Buffer.isBuffer(msg.payload)) {
                        msg.res._res.status(statusCode).jsonp(msg.payload);
                    } else {
                        if (msg.res._res.get('content-length') == null) {
                            var len;
                            if (msg.payload == null) {
                                len = 0;
                            } else if (Buffer.isBuffer(msg.payload)) {
                                len = msg.payload.length;
                            } else if (typeof msg.payload == "number") {
                                len = Buffer.byteLength(""+msg.payload);
                            } else {
                                len = Buffer.byteLength(msg.payload);
                            }
                            msg.res._res.set('content-length', len);
                        }


                        if (typeof msg.payload === "number") {
                            msg.payload = ""+msg.payload;
                        }
                        msg.res._res.status(statusCode).send(msg.payload);
                    }
                }
            } else {
                // ********************************** START CHANGE ***********************************
                if (node.streaming == false) {
                    node.warn(RED._("httpin.errors.no-response"));
                }
                else {
                    if(msg.payload && !msg.res) {
                        node.responses.forEach(function(resp) {
                            resp._res.write('--' + boundary);  
                            resp._res.write('\r\n');                                      
                            resp._res.write('Content-Type: image/jpeg');
                            resp._res.write('\r\n');  
                            resp._res.write('Content-length: ' + msg.payload.length);
                            resp._res.write('\r\n');
                            resp._res.write('\r\n');
                            resp._res.write(msg.payload);
                            resp._res.write('\r\n');
                            resp._res.write('\r\n');
                            //resp._res.flush();
                        });
                    }
                }
                // ********************************** END CHANGE ***********************************
            }
        });
    }
    RED.nodes.registerType("http response",HTTPOut);

All 'constructive' feedback is more than welcome!!!  I will start by listing a few remarks/questions:

• In the beginning I wanted to create a new httpstream node, but the amount of code added is not that much.  So I thought it would be better to add it to the existing httpout (httpresponse) node.  Is that a good decision or not ?
• I thought that I had to respond to the node's close event: when the node is (re)deployed the array of responses should be cleared?  But it seems to me that Expressjs cleans already the array, by using the handlers I have provided.  Is that a good decision or not ?
• Each image has it's own header section, where you have to specify the content-type, which is currently hardcoded to image/jpeg.  How should this be passed dynamically: by a contenttype field in the first message from the httpin node (4), or a contenttype field in every data message (6), or in the httpout config window, or ...
• At the end the msg.payload is being written to all the response objects:

                    if(msg.payload && !msg.res) {
                        node.responses.forEach(function(resp) {
                            ...
                            resp._res.write(msg.payload);
                            ...
                        });
                    }

However the write does only accept string or buffer types.  I'm not sure how I have to handle other types (dates, numbers, json objects ...).  Would be great if somebody could give some code/advise on this matter...  

• A second question about this latter code snippet:  I only send the msg.payload if !msg.res.   Reason is that the first message (4) contains the http request object in the payload.  Since the write doesn't support objects, an exception would be raised.  As a result nothing would be displayed in the browser, despite that the browser would be retrieving data all the time.  This problem was almost driving me nuts.  Is this a good check condition or is there a better solution?
  

I think this would be a great enhancement for Node-Red.  Currently I'm using a dashboard template node to display my camera images in an <img> tag.  However when more camera's are added (and when the frame rate is increasing), my browser starts hanging.  I have the impression that a single thread in the browser is handling all the data in the (single) webservice connection.  But when I use a series of <video> tags (with a src attribute referring to my stream urls), it seems to be working fine.  Looks like the browser is now using multiple threads to handle each separate stream...

Hopefully somebody got reading till here without suffering from a heart attack, and is still able to give feedback on my issues ;-)

Kind regards,

Bart Butenaers

[Nick O'Leary]

Hi Bart,

this is going to need some detailed consideration - you've clearly put a lot of thought into the current proposal.

There are a couple things that stick out to me.

The model assumes each request wants to join the stream at whatever point it's at - such as picking up the live stream from your web cam. That is one use of multipart streams, but we need to be careful to not get stuck in a corner when it comes to considering other use cases. Some consideration must be made for the use case where each request is expected to start receiving the stream from the start - such as playing back recorded footage.

Content-type needs to be configurable - the question is what the right approach is for exposing it. This is where fleshing out other use cases is a must.

Likewise all the other http response headers you currently hardcode - which need to be configurable by the user and how does the sit with the headers a user can already provide as part of the message passed to the response node (or, as of 0.17, set in the response node itself).

Nick

--
http://nodered.org
 
Join us on Slack to continue the conversation: http://nodered.org/slack
---
You received this message because you are subscribed to the Google Groups "Node-RED" group.
To unsubscribe from this group and stop receiving emails from it, send an email to node-red+unsubscribe@googlegroups.com.
To post to this group, send email to node...@googlegroups.com.
Visit this group at https://groups.google.com/group/node-red.
To view this discussion on the web, visit https://groups.google.com/d/msgid/node-red/835cf815-e8c1-434d-b57a-c11f19b54653%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[Bart Butenaers]

Evening Nick,

Damn, I thought this design was 100% Nick-proof ...  

But you are completely right that in some use cases all clients need to get the entire stream from the start (like playing recorded video).

Could following setup solve this?

Suppose the config view of the httpresponse node would have a dropdown:

The default option would be 'disabled' for existing flows.  When 'client independent' is selected, all msg.payload data would be send to ALL responses/clients (available in the node's array): this way the clients can hook in, in the middle of a stream that is already running.  When 'client dependent' is selected, the msg.payload will ONLY be send to the response object available in the msg.res field: this way the clients get there own private stream from the start.

Then we could do something like this:

When the first message (1) arrives in the httpresponse node, the headers (at stream start) are being written.  When the next messages (2) arrive, the data in the msg.payloads of those messages are written to the stream.  Or is this not correct, since the order of 1 and 2 can be reversed by the event queue mechanism : in that case the stream start headers would be written after the data has arrived ? 

And perhaps it is not a good practice to pass the response object in all the messages (msg.res), but instead it might perhaps be better to pass a simple response identifier ?  So that 1 sends a response object, while 2 passes only the response identifier (and the httpresponse node stores the responses in a map with the identifier as key) ...

Suddenly I have more questions than answers ;-(

Bart

[Bart Butenaers]

Nick,

I created a second version of my encoder, with support of my dropdown above.  Now clients can connect either hook in a stream that is already running, or they can get there own stream from the beginning.  This is indeed a great improvement that supports much more use cases.  Thank you very much for this useful insight !!!!!!!!!!!

But I'm struggling with your second remark about dynamically setting the http headers.  Because a multipart stream has two kinds of http headers:

HTTP/1.0 200 OK

Content-Type: multipart/x-mixed-replace;boundary=--myboundary

Connection: keep-alive

Expires: Fri, 01 Jan 1990 00:00:00 GMT

Cache-Control: no-cache, no-store, max-age=0, must-revalidate

Pragma: no-cache

--boundary
Content-Type: image/jpeg
Content-Length: [length of the content data bytes]

[content data bytes e.g. camera image]

--boundary
Content-Type: image/jpeg
Content-Length: [length of the content data bytes]

[content data bytes e.g. camera image]

...

At the start of the stream we have a global http header.  The content-type header field needs to contain the 'multipart' and 'boundary' stuff, and the other cache-related headers should also be available.  If somebody should remove these, the streaming simply doesn't work.  I could of course add these required headers by default to the 'headers' section on the config screen:

This way a user could change them (or add extra headers), but they are at least set by default.

Moreover each part of the multipart stream also contains a part http header, which also should be dynamically adjustable.  I assume the most easy way to implement this, is to display a second list (in case streaming is enabled):

Moreover, both types of headers should be dynamically adjustable using the input message.  I could a msg.partheaders field beside to the already existing msg.headers , but I think that would become confusing: indeed the global headers can only be applied once (at the start of the stream).  When applied afterwards, following error would occur: 

Error: Can't set headers after they are sent.

This means that in the 2nd and next messages these msg.headers field should be ignored by the node (which might be confusing: users set headers and they are not used).  Therefore it might be better to reuse the existing msg.headers field both for global headers and part headers. As a result the first message would contain the global headers (and the payload will be ignored, i.e. not send), and from the 2nd message it contains the part headers (and the payload contains data that will be send).

Everybody is welcome to join this discussion !

Kind regards,

Bart

[Bart Butenaers]

Hi everybody,

At first sight this was only a small enhancement of the http-response node, so I started this discussion as a prerequisite for a future pull-request.  However due to the (very useful!) feedback of Nick, it became very obvious that there were rather few similarities between the http-response node and the http-streaming node I had in mind...

So I think it would become very confusing for users that would like to use the http-response node for standard (non-streaming) stuff.

Therefore I have put all my streaming code in a new contribution : node-red-contrib-multipart-stream-encoder

In short: it allows you to create a dashboard with N streaming camera views, without the browser freezing (too soon)...

A lot of explanation had been added to the readme page, so people have all the info at a single location.

I have not published it on NPM yet, so you can install it like this:

npm install https://github.com/bartbutenaers/node-red-contrib-multipart-stream-encoder.git

Hopefully some users can find some time to experiment with it, before I publish it.

As usual, all 'constructive' feedback is more than welcome !!!!!!!!!

Thanks in advance,

Bart

[Bart Butenaers]

Hi everybody,

I have just published my contribution to npm: https://www.npmjs.com/package/node-red-contrib-multipart-stream-encoder

Since there was (except from Nick) nobody interested in extra functionalities, I was only able to add some of my own ideas:

• When connections are closed, an output message will be generated (so the flow can respond to this by stop sending data)
• By sending a msg.close = true you can end connections yourself 
• Adaptive streaming : when the node detects that the stream is consuming less data, it will start skipping input messages.  And otherwise when the stream starts consuming more data, it will start sending more message payloads to the stream.

The config view now has a separate tables for global headers and part headers to be able to implement Nick's proposal (to make the headers adjustable).  These tables are filled with default headers that are needed for a multipart stream;

Any feedback is still welcome (here or as an issue on Github)...

Bart