Merge remote-tracking branch 'upstream/master'

nodes/core/io/10-mqtt.html

@@ -30,8 +30,9 @@
 </script>
 
 <script type="text/x-red" data-help-name="mqtt in">
-	<p>MQTT input node. Connects to the specified broker and subscribes to the specified topic. The topic may contain MQTT wildcards.</p>
+	<p>MQTT input node. Connects to a broker and subscribes to the specified topic. The topic may contain MQTT wildcards.</p>
-	<p>Outputs an object called <b>msg</b> containing <b>msg.topic, msg.payload, msg.qos</b> and <b>msg.retain</b>. <b>msg.payload</b> is a String.</p>
+	<p>Outputs an object called <b>msg</b> containing <b>msg.topic, msg.payload, msg.qos</b> and <b>msg.retain</b>.</p>
+	<p><b>msg.payload</b> is a String.</p>
 </script>
 
 <script type="text/javascript">
@@ -105,6 +106,18 @@
         <label for="node-config-input-port" style="margin-left: 10px; width: 35px; "> Port</label>
         <input type="text" id="node-config-input-port" placeholder="Port" style="width:45px">
     </div>
+    <div class="form-row">
+        <label for="node-config-input-clientid"><i class="icon-tag"></i> Client ID</label>
+        <input type="text" id="node-config-input-clientid" placeholder="Leave blank for auto generated">
+    </div>
+    <div class="form-row">
+        <label for="node-config-input-user"><i class="icon-user"></i> Username</label>
+        <input type="text" id="node-config-input-user">
+    </div>
+    <div class="form-row">
+        <label for="node-config-input-pass"><i class="icon-lock"></i> Password</label>
+        <input type="password" id="node-config-input-pass">
+    </div>
 </script>
 
 <script type="text/javascript">
@@ -112,10 +125,49 @@
         category: 'config',
         defaults: {
             broker: {value:"localhost",required:true},
-            port: {value:1883,required:true,validate:RED.validators.number()}
+            port: {value:1883,required:true,validate:RED.validators.number()},
+            clientid: { value:"" }
+            //user -> credentials
+            //pass -> credentials
+            
         },
         label: function() {
             return this.broker+":"+this.port;
+        },
+        oneditprepare: function() {
+            $.getJSON('mqtt-broker/'+this.id,function(data) {
+                if (data.user) {
+                    $('#node-config-input-user').val(data.user);
+                }
+                if (data.hasPassword) {
+                    $('#node-config-input-pass').val('__PWRD__');
+                } else {
+                    $('#node-config-input-pass').val('');
+                }
+                
+            });
+        },
+        oneditsave: function() {
+            var newUser = $('#node-config-input-user').val();
+            var newPass = $('#node-config-input-pass').val();
+            var credentials = {};
+            credentials.user = newUser;
+            if (newPass != '__PWRD__') {
+                credentials.password = newPass;
+            }
+            $.ajax({
+                url: 'mqtt-broker/'+this.id,
+                type: 'POST',
+                data: credentials,
+                success:function(result){}
+            });
+        },
+        ondelete: function() {
+            $.ajax({
+                url: 'mqtt-broker/'+this.id,
+                type: 'DELETE',
+                success: function(result) {}
+            });
         }
     });
 </script>

nodes/core/io/10-mqtt.js

@@ -22,9 +22,54 @@ function MQTTBrokerNode(n) {
     RED.nodes.createNode(this,n);
     this.broker = n.broker;
     this.port = n.port;
+    this.clientid = n.clientid;
+    var credentials = RED.nodes.getCredentials(n.id);
+    if (credentials) {
+        this.username = credentials.user;
+        this.password = credentials.password;
+    }  
 }
 RED.nodes.registerType("mqtt-broker",MQTTBrokerNode);
 
+var querystring = require('querystring');
+
+RED.app.get('/mqtt-broker/:id',function(req,res) {
+    var credentials = RED.nodes.getCredentials(req.params.id);
+    if (credentials) {
+        res.send(JSON.stringify({user:credentials.user,hasPassword:(credentials.password&&credentials.password!="")}));
+    } else {
+        res.send(JSON.stringify({}));
+    }
+});
+
+RED.app.delete('/mqtt-broker/:id',function(req,res) {
+    RED.nodes.deleteCredentials(req.params.id);
+    res.send(200);
+});
+
+RED.app.post('/mqtt-broker/:id',function(req,res) {
+    var body = "";
+    req.on('data', function(chunk) {
+        body+=chunk;
+    });
+    req.on('end', function(){
+        var newCreds = querystring.parse(body);
+        var credentials = RED.nodes.getCredentials(req.params.id)||{};
+        if (newCreds.user == null || newCreds.user == "") {
+            delete credentials.user;
+        } else {
+            credentials.user = newCreds.user;
+        }
+        if (newCreds.password == "") {
+            delete credentials.password;
+        } else {
+            credentials.password = newCreds.password||credentials.password;
+        }
+        RED.nodes.addCredentials(req.params.id,credentials);
+        res.send(200);
+    });
+});
+
 
 function MQTTInNode(n) {
     RED.nodes.createNode(this,n);
@@ -32,7 +77,7 @@ function MQTTInNode(n) {
     this.broker = n.broker;
     this.brokerConfig = RED.nodes.getNode(this.broker);
     if (this.brokerConfig) {
-        this.client = connectionPool.get(this.brokerConfig.broker,this.brokerConfig.port);
+        this.client = connectionPool.get(this.brokerConfig.broker,this.brokerConfig.port,this.brokerConfig.clientid,this.brokerConfig.username,this.brokerConfig.password);
         var node = this;
         this.client.subscribe(this.topic,2,function(topic,payload,qos,retain) {
                 var msg = {topic:topic,payload:payload,qos:qos,retain:retain};
@@ -65,7 +110,7 @@ function MQTTOutNode(n) {
     this.brokerConfig = RED.nodes.getNode(this.broker);
 
     if (this.brokerConfig) {
-        this.client = connectionPool.get(this.brokerConfig.broker,this.brokerConfig.port);
+        this.client = connectionPool.get(this.brokerConfig.broker,this.brokerConfig.port,this.brokerConfig.clientid,this.brokerConfig.username,this.brokerConfig.password);
         this.on("input",function(msg) {
             if (msg != null) {
                 if (this.topic) {

nodes/core/io/22-websocket.js

@@ -43,6 +43,8 @@ function WebSocketListenerNode(n) {
         }
     }
 
+    node._clients = {};
+    
     RED.server.addListener('newListener',storeListener);
 
     // Create a WebSocket Server
@@ -53,8 +55,14 @@ function WebSocketListenerNode(n) {
     RED.server.removeListener('newListener',storeListener);
 
     node.server.on('connection', function(socket){
+        var id = (1+Math.random()*4294967295).toString(16);
+        node._clients[id] = socket;
+        
+        socket.on('close',function() {
+            delete node._clients[id];
+        });
         socket.on('message',function(data,flags){
-            node.handleEvent(socket,'message',data,flags);
+            node.handleEvent(id,socket,'message',data,flags);
         });
     });
 
@@ -80,9 +88,9 @@ WebSocketListenerNode.prototype.registerInputNode = function(/*Node*/handler){
     this._inputNodes.push(handler);
 }
 
-WebSocketListenerNode.prototype.handleEvent = function(/*socket*/socket,/*String*/event,/*Object*/data,/*Object*/flags){
+WebSocketListenerNode.prototype.handleEvent = function(id,/*socket*/socket,/*String*/event,/*Object*/data,/*Object*/flags){
     for (var i = 0; i < this._inputNodes.length; i++) {
-        this._inputNodes[i].send({payload:data});
+        this._inputNodes[i].send({session:{type:"websocket",id:id},payload:data});
     };
 }
 
@@ -92,6 +100,13 @@ WebSocketListenerNode.prototype.broadcast = function(data){
     };
 }
 
+WebSocketListenerNode.prototype.send = function(id,data){
+    var session = this._clients[id];
+    if (session) {
+        session.send(data);
+    }
+}
+
 function WebSocketInNode(n) {
     RED.nodes.createNode(this,n);
     this.server = n.server;
@@ -114,11 +129,23 @@ function WebSocketOutNode(n) {
         this.error("Missing server configuration");
     }
     this.on("input", function(msg) {
-        node.serverConfig.broadcast(msg.payload,function(error){
+        var payload = msg.payload;
+        if (Buffer.isBuffer(payload)) {
+            payload = payload.toString();
+        } else if (typeof payload === "object") {
+            payload = JSON.stringify(payload);
+        } else if (typeof payload !== "string") {
+            payload = ""+payload;
+        }
+        if (msg.session && msg.session.type == "websocket") {
+            node.serverConfig.send(msg.session.id,payload);
+        } else {
+            node.serverConfig.broadcast(payload,function(error){
                 if(!!error){
                     node.warn("An error occurred while sending:" + inspect(error));
                 }
             });
+        }
     });
 }
 RED.nodes.registerType("websocket out",WebSocketOutNode);

nodes/core/io/lib/mqttConnectionPool.js

@@ -25,13 +25,16 @@ function matchTopic(ts,t) {
 }
 
 module.exports = {
-    get: function(broker,port) {
+    get: function(broker,port,clientid,username,password) {
-        var id = broker+":"+port;
+        var id = "["+(username||"")+":"+(password||"")+"]["+(clientid||"")+"]@"+broker+":"+port;
         if (!connections[id]) {
             connections[id] = function() {
                 var client = mqtt.createClient(port,broker);
                 client.setMaxListeners(0);
-                var options = {keepalive:15,clientId:'mqtt_' + (1+Math.random()*4294967295).toString(16)};
+                var options = {keepalive:15};
+                options.clientId = clientid || 'mqtt_' + (1+Math.random()*4294967295).toString(16);
+                options.username = username;
+                options.password = password;
                 var queue = [];
                 var subscriptions = [];
                 var connecting = false;

nodes/core/logic/10-switch.html

@@ -60,7 +60,7 @@
         outputs: 1,
         icon: "switch.png",
         label: function() {
-            return this.name;
+            return this.name||"switch";
         },
         oneditprepare: function() {
 

nodes/core/logic/15-change.html

@@ -0,0 +1,111 @@
+<!--
+  Copyright 2013 IBM Corp.
+
+  Licensed under the Apache License, Version 2.0 (the "License");
+  you may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+-->
+
+<script type="text/x-red" data-template-name="change">
+    <div>
+        <select id="node-input-action" style="width:95%; margin-right:5px;">
+            <option value="change">Change the value of the property</option>
+            <!-- <option value="replace">Replace value of property</option> -->
+            <option value="replace">Add or replace property</option>
+            <option value="delete">Delete property</option>
+        </select>
+    </div>
+    <div class="form-row" style="padding-top:10px;" id="node-prop1-row">
+        <label id="node-input-todo">called</label>msg.<input type="text" id="node-input-property" style="width: 63%;"/>
+    </div>
+    <div class="form-row" id="node-from-row">
+        <label id="node-input-f">if it contains</label>
+        <input type="text" id="node-input-from" placeholder="this"/>
+    </div>
+    <div class="form-row" id="node-to-row">
+        <label id="node-input-t">replace with</label>
+        <input type="text" id="node-input-to" placeholder="that"/>
+    </div>
+    <br/>
+    <div class="form-row">
+        <label for="node-input-name"><i class="icon-tag"></i> Name</label>
+        <input type="text" id="node-input-name" placeholder="Name">
+    </div>
+    <div class="form-tips" id="node-tip"></div>
+</script>
+
+<script type="text/x-red" data-help-name="change">
+    <p>A simple function node to change, replace, add or delete properties of a message.</p>
+    <p>When a message arrives, the selected property is modified by the defined rules.
+    The message is then sent to the output.</p>
+    <p><b>Note:</b> Replace only operates on <b>strings</b>. Anything else will be passed straight through.</p>
+</script>
+
+<script type="text/javascript">
+    RED.nodes.registerType('change', {
+        color: "#E2D96E",
+        category: 'function',
+        defaults: {
+            action: {value:"change",required:true},
+            property: {value:"payload"},
+            from: {value:""},
+            to: {value:""},
+            name: {value:""}
+        },
+        inputs: 1,
+        outputs: 1,
+        icon: "swap.png",
+        label: function() {
+            return this.name || this.action || "change";
+        },
+        labelStyle: function() {
+            return this.name ? "node_label_italic" : "";
+        },
+        oneditprepare: function() {
+            $("#node-input-action").change( function() {
+                var a = $("#node-input-action").val();
+                if (a === "replace") {
+                    $("#node-input-todo").html("called");
+                    //$("#node-input-f").html("name");
+                    $("#node-input-t").html("with value");
+                    $("#node-from-row").hide();
+                    $("#node-to-row").show();
+                    $("#node-tip").html("Tip: expects a new property name and either a fixed value OR the full name of another msg.property eg: msg.sentiment.score");
+                }
+                if (a === "delete") {
+                    $("#node-input-todo").html("called");
+                    //$("#node-input-f").html("called");
+                    //$("#node-input-t").html("to");
+                    $("#node-from-row").hide();
+                    $("#node-to-row").hide();
+                    $("#node-tip").html("Tip: deletes the named property and all sub-properties");
+                }
+                if (a === "change") {
+                    $("#node-input-todo").html("called");
+                    $("#node-input-f").html("If it contains");
+                    $("#node-input-t").html("replace with");
+                    $("#node-from-row").show();
+                    $("#node-to-row").show();
+                    $("#node-tip").html("Tip: <i>if it contains</i> can be a regex, likewise <i>replace with</i> can accept regex results. Only works on strings.");
+                }
+                //if (a === "replace") {
+                 //   $("#node-input-todo").html("called");
+                 //   //$("#node-input-f").html("with");
+                 //   $("#node-input-t").html("with");
+                 //   $("#node-from-row").hide();
+                 //   $("#node-to-row").show();
+                 //   $("#node-tip").html("Tip: accepts either a fixed value OR the full name of another msg.property eg: msg.sentiment.score");
+                //}
+            });
+            $("#node-input-action").change();
+        }
+    });
+</script>

nodes/core/logic/15-change.js

@@ -0,0 +1,66 @@
+/**
+ * Copyright 2013 IBM Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ **/
+
+var RED = require(process.env.NODE_RED_HOME + "/red/red");
+
+function ChangeNode(n) {
+    RED.nodes.createNode(this, n);
+    this.action = n.action;
+    this.property = n.property || "";
+    this.from = n.from || " ";
+    this.to = n.to || " ";
+    var node = this;
+
+    var makeNew = function( stem, path, value ) {
+        var lastPart = (arguments.length === 3) ? path.pop() : false;
+        for (var i = 0; i < path.length; i++) {
+            stem = stem[path[i]] = stem[path[i]] || {};
+        }
+        if (lastPart) { stem = stem[lastPart] = value; }
+        return stem;
+    };
+
+    this.on('input', function (msg) {
+        if (node.action == "change") {
+            node.re = new RegExp(this.from, "g");
+            if (typeof msg[node.property] === "string") {
+                msg[node.property] = (msg[node.property]).replace(node.re, node.to);
+            }
+        }
+        //else if (node.action == "replace") {
+            //if (node.to.indexOf("msg.") == 0) {
+                //msg[node.property] = eval(node.to);
+            //}
+            //else {
+                //msg[node.property] = node.to;
+            //}
+        //}
+        else if (node.action == "replace") {
+            if (node.to.indexOf("msg.") == 0) {
+                makeNew( msg, node.property.split("."), eval(node.to) );
+            }
+            else {
+                makeNew( msg, node.property.split("."), node.to );
+            }
+            //makeNew( msg, node.property.split("."), node.to );
+        }
+        else if (node.action == "delete") {
+            delete(msg[node.property]);
+        }
+        node.send(msg);
+    });
+}
+RED.nodes.registerType("change", ChangeNode);

nodes/core/social/27-twitter.html

@@ -129,11 +129,19 @@
 </script>
 
 <script type="text/x-red" data-help-name="twitter in">
-    <p>Twitter input node. Watches either the public or the user's stream for tweets containing the configured search term.</p>
+    <p>Twitter input node. Can be used to search either:
+    <ul><li>the public or a user's stream for tweets containing the configured search term</li>
+        <li>all tweets by specific users</li>
+        <li>direct messages received by the authenticated user</li>
+    </ul></p>
     <p>Use space for <i>and</i> and comma , for <i>or</i> when searching for multiple terms.</p>
     <p>Sets the <b>msg.topic</b> to <i>tweets/</i> and then appends the senders screen name.</p>
     <p>Sets <b>msg.location</b> to the tweeters location if known.</p>
     <p>Sets <b>msg.tweet</b> to the full tweet object as documented by <a href="https://dev.twitter.com/docs/platform-objects/tweets">Twitter</a>.
+    <p><b>Note:</b> when set to a specific user's tweets, or your direct messages, the node is subject to
+      Twitter's API rate limiting. If you deploy the flows multiple times within a 15 minute window, you may
+      exceed the limit and will see errors from the node. These errors will clear when the current 15 minute window
+      passes.</p>
 </script>
 
 <script type="text/javascript">
@@ -142,7 +150,7 @@
         color:"#C0DEED",
         defaults: {
             twitter: {type:"twitter-credentials",required:true},
-            tags: {value:"",required:true},
+            tags: {value:"",validate:function(v) { return this.user == "dm" || v.length > 0;}},
             user: {value:"false",required:true},
             name: {value:""},
             topic: {value:"tweets"}
@@ -151,7 +159,16 @@
         outputs:1,
         icon: "twitter.png",
         label: function() {
-            return this.name||this.tags;
+            if (this.name) {
+                return this.name;
+            }
+            if (this.user == "dm") {
+                var user = RED.nodes.node(this.twitter);
+                return (user?user.label()+" ":"")+"DMs";
+            } else if (this.user == "user") {
+                return this.tags+" tweets";
+            }
+            return this.tags;
         },
         labelStyle: function() {
             return this.name?"node_label_italic":"";

nodes/core/social/27-twitter.js

@@ -43,6 +43,14 @@ function TwitterInNode(n) {
             access_token_secret: credentials.access_token_secret
         });
 
+        
+        //setInterval(function() {
+        //        twit.get("/application/rate_limit_status.json",null,function(err,cb) {
+        //                console.log("direct_messages:",cb["resources"]["direct_messages"]);
+        //        });
+        //        
+        //},10000);
+        
         var node = this;
         if (this.user === "user") {
             node.poll_ids = [];
@@ -118,10 +126,7 @@ function TwitterInNode(n) {
                             if (cb) {
                                 for (var t=cb.length-1;t>=0;t-=1) {
                                     var tweet = cb[t];
-                                    var where = tweet.user.location||"";
+                                    var msg = { topic:node.topic+"/"+tweet.sender.screen_name, payload:tweet.text, tweet:tweet };
-                                    var la = tweet.lang || tweet.user.lang;
-                                    //console.log(tweet.user.location,"=>",tweet.user.screen_name,"=>",pay);
-                                    var msg = { topic:node.topic+"/"+tweet.user.screen_name, payload:tweet.text, location:where, lang:la, tweet:tweet };
                                     node.send(msg);
                                     if (t == 0) {
                                         node.since_id = tweet.id_str;
@@ -132,7 +137,7 @@ function TwitterInNode(n) {
                                 node.error(err);
                             }
                         });
-                },65000));
+                },120000));
             });
             
         } else if (this.tags !== "") {

nodes/core/social/32-feedparse.js

@@ -35,23 +35,19 @@ function FeedParseNode(n) {
                         node.error(error);
                 })
                 .on('meta', function (meta) {})
-                .on('article', function (article) {
+                .on('readable', function () {
+                        var stream = this, article;
+                        while (article = stream.read()) {
                             if (!(article.guid in node.seen) || ( node.seen[article.guid] != 0 && node.seen[article.guid] != article.date.getTime())) {
                                 node.seen[article.guid] = article.date?article.date.getTime():0;
                                 var msg = {
                                     topic:article.origlink||article.link,
                                     payload: article.description,
-                                article: {
+                                    article: article
-                                    summary:article.summary,
-                                    link:article.link,
-                                    date: article.date,
-                                    pubdate: article.pubdate,
-                                    author: article.author,
-                                    guid: article.guid,
-                                }
                                 };
                                 node.send(msg);
                             }
+                        }
                 })
                 .on('end', function () {
                 });

nodes/core/social/61-imap.html

@@ -27,10 +27,11 @@
 
 <script type="text/x-red" data-help-name="imap">
     <p>Repeatedly gets a <b>single email</b> from an IMAP server and forwards on as a msg if not already seen.</p>
-	<p>The subject is loaded into <b>msg.topic</b> and <b>msg.payload</b> is the body text. <b>msg.from</b> is also set if you need it.</p>
+    <p>The subject is loaded into <b>msg.topic</b> and <b>msg.payload</b> is the plain text body.
+       If there is text/html then that is returned in <b>msg.html</b>. <b>msg.from</b> and <b>msg.date</b> are also set if you need them.</p>
     <p>Uses the imap module - you also need to pre-configure your email settings in a file emailkeys.js as per below.</p>
     <p><pre>module.exports = { service: "Gmail", user: "blahblah@gmail.com", pass: "password", server: "imap.gmail.com", port: "993" }</pre></p>
-	<p>This <b>must</b> be located in the diectory above node-red.</p>
+    <p>This <b>must</b> be located in the directory above node-red.</p>
     <p><b>Note:</b> this node <i>only</i> gets the most recent single email from the inbox, so set the repeat (polling) time appropriately.</p>
 </script>
 

nodes/core/social/61-imap.js

@@ -17,12 +17,12 @@
 var RED = require(process.env.NODE_RED_HOME+"/red/red");
 var Imap = require('imap');
 var util = require('util');
-var oldmail = {};
 
 try {
     var emailkey = RED.settings.email || require(process.env.NODE_RED_HOME+"/../emailkeys.js");
 } catch (err) {
-	throw new Error("Failed to load Email credentials");
+    util.log("[imap] : Failed to load Email credentials");
+    return;
 }
 
 var imap = new Imap({
@@ -30,79 +30,98 @@ var imap = new Imap({
     password: emailkey.pass,
     host: emailkey.server||"imap.gmail.com",
     port: emailkey.port||"993",
-	secure: true
+    tls: true,
+    tlsOptions: { rejectUnauthorized: false }
 });
 
-function fail(err) {
-	util.log('[imap] : ' + err);
-}
-
 function openInbox(cb) {
-	imap.connect(function(err) {
-		if (err) fail(err);
     imap.openBox('INBOX', true, cb);
-	});
 }
 
 function ImapNode(n) {
     RED.nodes.createNode(this,n);
     this.name = n.name;
-	this.repeat = n.repeat * 1000;
+    this.repeat = n.repeat * 1000 || 300000;
     var node = this;
     this.interval_id = null;
+    var oldmail = {};
 
-	if (this.repeat && !isNaN(this.repeat) && this.repeat > 0) {
+    if (!isNaN(this.repeat) && this.repeat > 0) {
-		this.log("repeat = "+this.repeat);
+        node.log("repeat = "+this.repeat);
         this.interval_id = setInterval( function() {
             node.emit("input",{});
         }, this.repeat );
     }
 
     this.on("input", function(msg) {
-		openInbox(function(err, mailbox) {
+        imap.once('ready', function() {
-		  if (err) fail(err);
-		  imap.seq.fetch(mailbox.messages.total + ':*', { struct: false },
-			{ headers: ['from', 'subject'],
-			  body: true,
-			  cb: function(fetch) {
-				fetch.on('message', function(msg) {
-				  //node.log('Saw message no. ' + msg.seqno);
             var pay = {};
-				  var body = '';
+            openInbox(function(err, box) {
-				  msg.on('headers', function(hdrs) {
+                //if (err) throw err;
-					pay.from = hdrs.from[0];
+                var f = imap.seq.fetch(box.messages.total + ':*', { bodies: ['HEADER.FIELDS (FROM SUBJECT DATE)','TEXT'] });
-					pay.topic = hdrs.subject[0];
+                f.on('message', function(msg, seqno) {
+                    node.log('message: #'+ seqno);
+                    var prefix = '(#' + seqno + ') ';
+                    msg.on('body', function(stream, info) {
+                        var buffer = '';
+                        stream.on('data', function(chunk) {
+                            buffer += chunk.toString('utf8');
+                        });
+                        stream.on('end', function() {
+                            if (info.which !== 'TEXT') {
+                                pay.from = Imap.parseHeader(buffer).from[0];
+                                pay.topic = Imap.parseHeader(buffer).subject[0];
+                                pay.date = Imap.parseHeader(buffer).date[0];
+                            } else {
+                                var parts = buffer.split("Content-Type");
+                                for (var p in parts) {
+                                    if (parts[p].indexOf("text/plain") >= 0) {
+                                        pay.payload = parts[p].split("\n").slice(1,-2).join("\n").trim();
+                                    }
+                                    if (parts[p].indexOf("text/html") >= 0) {
+                                        pay.html = parts[p].split("\n").slice(1,-2).join("\n").trim();
+                                    }
+                                }
+                                //pay.body = buffer;
+                            }
                         });
-				  msg.on('data', function(chunk) {
-					body += chunk.toString('utf8');
                     });
                     msg.on('end', function() {
-					pay.payload = body;
+                        //node.log('Finished: '+prefix);
-					if ((pay.topic !== oldmail.topic)|(pay.payload !== oldmail.payload)) {
+                    });
-						oldmail = pay;
+                });
-						//node.log("From: "+pay.from);
+                f.on('error', function(err) {
-						node.log("Subj: "+pay.topic);
+                    node.warn('fetch error: ' + err);
-						//node.log("Body: "+pay.payload);
+                });
+                f.on('end', function() {
+                    if (JSON.stringify(pay) !== oldmail) {
                         node.send(pay);
+                        oldmail = JSON.stringify(pay);
+                        node.log('sent new message: '+pay.topic);
                     }
+                    else { node.log('duplicate not sent: '+pay.topic); }
+                    imap.end();
                 });
             });
-			  }
+        });
-			}, function(err) {
+        imap.connect();
-				if (err) node.log("Err : "+err);
-				//node.log("Done fetching messages.");
-				imap.logout();
-			}
-		  );
     });
 
+    imap.on('error', function(err) {
+        util.log(err);
+    });
+
+    this.on("error", function(err) {
+        node.log("error: ",err);
     });
 
     this.on("close", function() {
         if (this.interval_id != null) {
             clearInterval(this.interval_id);
         }
+        imap.destroy();
     });
-}
 
+    node.emit("input",{});
+}
 RED.nodes.registerType("imap",ImapNode);

red.js

@@ -18,12 +18,30 @@ var https = require('https');
 var util = require("util");
 var express = require("express");
 var crypto = require("crypto");
-var settings = require("./settings");
 var RED = require("./red/red.js");
 
 var server;
 var app = express();
 
+var settingsFile = "./settings";
+var flowFile;
+
+for (var argp = 2;argp < process.argv.length;argp+=1) {
+    var v = process.argv[argp];
+    if (v == "--settings" || v == "-s") {
+        if (argp+1 == process.argv.length) {
+            console.log("Missing argument to --settings");
+            return;
+        }
+        argp++;
+        settingsFile = process.argv[argp];
+    } else {
+        flowFile = v;
+    }
+}
+
+var settings = require(settingsFile);
+
 if (settings.https) {
     server = https.createServer(settings.https,function(req,res){app(req,res);});
 } else {
@@ -49,7 +67,7 @@ if (settings.httpAuth) {
     );
 }
 
-settings.flowFile = process.argv[2] || settings.flowFile;
+settings.flowFile = flowFile || settings.flowFile;
 
 var red = RED.init(server,settings);
 app.use(settings.httpRoot,red);

red/nodes.js

@@ -155,28 +155,36 @@ Node.prototype.send = function(msg) {
                 if (!util.isArray(msg[i])) {
                     msgs = [msg[i]];
                 }
-                if (wires.length == 1) {
+                //if (wires.length == 1) {
-                    // Single recipient, don't need to clone the message
+                //    // Single recipient, don't need to clone the message
-                    var node = registry.get(wires[0]);
+                //    var node = registry.get(wires[0]);
-                    if (node) {
+                //    if (node) {
-                        for (var k in msgs) {
+                //        for (var k in msgs) {
-                            var mm = msgs[k];
+                //            var mm = msgs[k];
-                            node.receive(mm);
+                //            node.receive(mm);
-                        }
+                //        }
-                    }
+                //    }
-                } else {
+                //} else {
                     // Multiple recipients, must send message copies
                     for (var j in wires) {
                         var node = registry.get(wires[j]);
                         if (node) {
                             for (var k in msgs) {
                                 var mm = msgs[k];
+                                // Temporary fix for #97
+                                // TODO: remove this http-node-specific fix somehow
+                                var req = mm.req;
+                                var res = mm.res;
+                                mm.req = null;
+                                mm.res = null;
                                 var m = clone(mm);
+                                m.req = req;
+                                m.res = res;
                                 node.receive(m);
                             }
                         }
                     }
-                }
+                //}
             }
         }
     }