1
0
mirror of https://github.com/node-red/node-red.git synced 2023-10-10 13:36:53 +02:00
node-red/nodes/core/storage/65-redisout.js

112 lines
3.9 KiB
JavaScript
Raw Normal View History

2013-09-05 16:02:48 +02:00
/**
* 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.
**/
2014-05-04 00:32:04 +02:00
module.exports = function(RED) {
2014-07-17 11:25:52 +02:00
"use strict";
2014-05-04 00:32:04 +02:00
var util = require("util");
var redis = require("redis");
2014-07-17 11:25:52 +02:00
2014-05-04 00:32:04 +02:00
var hashFieldRE = /^([^=]+)=(.*)$/;
2014-07-17 11:25:52 +02:00
2014-05-04 00:32:04 +02:00
var redisConnectionPool = function() {
var connections = {};
var obj = {
get: function(host,port) {
var id = host+":"+port;
if (!connections[id]) {
connections[id] = redis.createClient(port,host);
connections[id].on("error",function(err) {
util.log("[redis] "+err);
});
connections[id].on("connect",function() {
util.log("[redis] connected to "+host+":"+port);
});
connections[id]._id = id;
connections[id]._nodeCount = 0;
}
connections[id]._nodeCount += 1;
return connections[id];
},
close: function(connection) {
connection._nodeCount -= 1;
2014-07-17 11:25:52 +02:00
if (connection._nodeCount === 0) {
2014-05-04 00:32:04 +02:00
if (connection) {
clearTimeout(connection.retry_timer);
connection.end();
}
delete connections[connection._id];
2013-09-05 16:02:48 +02:00
}
}
2014-05-04 00:32:04 +02:00
};
return obj;
}();
2014-07-17 11:25:52 +02:00
2014-05-04 00:32:04 +02:00
function RedisOutNode(n) {
RED.nodes.createNode(this,n);
this.port = n.port||"6379";
this.hostname = n.hostname||"127.0.0.1";
this.key = n.key;
this.structtype = n.structtype;
2014-07-17 11:25:52 +02:00
2014-05-04 00:32:04 +02:00
this.client = redisConnectionPool.get(this.hostname,this.port);
if (this.client.connected) {
this.status({fill:"green",shape:"dot",text:"connected"});
} else {
this.status({fill:"red",shape:"ring",text:"disconnected"},true);
}
var node = this;
this.client.on("end", function() {
node.status({fill:"red",shape:"ring",text:"disconnected"});
});
this.client.on("connect", function() {
node.status({fill:"green",shape:"dot",text:"connected"});
});
2014-05-04 00:32:04 +02:00
this.on("input", function(msg) {
2014-09-08 22:10:06 +02:00
var k = this.key || msg.topic;
if (k) {
if (this.structtype == "string") {
this.client.set(k,RED.util.ensureString(msg.payload));
} else if (this.structtype == "hash") {
if (typeof msg.payload == "object") {
this.client.hmset(k,msg.payload);
2014-05-04 00:32:04 +02:00
} else {
var r = hashFieldRE.exec(msg.payload);
if (r) {
this.client.hset(k,r[1],r[2]);
} else {
this.warn("Invalid payload for redis hash");
}
2013-09-05 16:02:48 +02:00
}
2014-09-08 22:10:06 +02:00
} else if (this.structtype == "set") {
this.client.sadd(k,msg.payload);
} else if (this.structtype == "list") {
this.client.rpush(k,msg.payload);
2013-09-05 16:02:48 +02:00
}
2014-09-08 22:10:06 +02:00
} else {
this.warn("No key or topic set");
}
2014-05-04 00:32:04 +02:00
});
2014-07-17 11:25:52 +02:00
this.on("close", function() {
redisConnectionPool.close(node.client);
});
2014-05-04 00:32:04 +02:00
}
RED.nodes.registerType("redis out",RedisOutNode);
2013-09-05 16:02:48 +02:00
}