dgbuilder/core_nodes/storage/65-redisout.js

   1 /**
   2  * Copyright 2013 IBM Corp.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  * http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  **/
  16
  17 module.exports = function(RED) {
  18     "use strict";
  19     var util = require("util");
  20     var redis = require("redis");
  21
  22     var hashFieldRE = /^([^=]+)=(.*)$/;
  23
  24     var redisConnectionPool = function() {
  25         var connections = {};
  26         var obj = {
  27             get: function(host,port) {
  28                 var id = host+":"+port;
  29                 if (!connections[id]) {
  30                     connections[id] = redis.createClient(port,host);
  31                     connections[id].on("error",function(err) {
  32                             util.log("[redis] "+err);
  33                     });
  34                     connections[id].on("connect",function() {
  35                             util.log("[redis] connected to "+host+":"+port);
  36                     });
  37                     connections[id]._id = id;
  38                     connections[id]._nodeCount = 0;
  39                 }
  40                 connections[id]._nodeCount += 1;
  41                 return connections[id];
  42             },
  43             close: function(connection) {
  44                 connection._nodeCount -= 1;
  45                 if (connection._nodeCount === 0) {
  46                     if (connection) {
  47                         clearTimeout(connection.retry_timer);
  48                         connection.end();
  49                     }
  50                     delete connections[connection._id];
  51                 }
  52             }
  53         };
  54         return obj;
  55     }();
  56
  57
  58     function RedisOutNode(n) {
  59         RED.nodes.createNode(this,n);
  60         this.port = n.port||"6379";
  61         this.hostname = n.hostname||"127.0.0.1";
  62         this.key = n.key;
  63         this.structtype = n.structtype;
  64
  65         this.client = redisConnectionPool.get(this.hostname,this.port);
  66
  67         if (this.client.connected) {
  68             this.status({fill:"green",shape:"dot",text:"connected"});
  69         } else {
  70             this.status({fill:"red",shape:"ring",text:"disconnected"},true);
  71         }
  72
  73         var node = this;
  74         this.client.on("end", function() {
  75             node.status({fill:"red",shape:"ring",text:"disconnected"});
  76         });
  77         this.client.on("connect", function() {
  78             node.status({fill:"green",shape:"dot",text:"connected"});
  79         });
  80
  81         this.on("input", function(msg) {
  82             var k = this.key || msg.topic;
  83             if (k) {
  84                 if (this.structtype == "string") {
  85                     this.client.set(k,RED.util.ensureString(msg.payload));
  86                 } else if (this.structtype == "hash") {
  87                     var r = hashFieldRE.exec(msg.payload);
  88                     if (r) {
  89                         this.client.hset(k,r[1],r[2]);
  90                     } else {
  91                         this.warn("Invalid payload for redis hash");
  92                     }
  93                 } else if (this.structtype == "set") {
  94                     this.client.sadd(k,msg.payload);
  95                 } else if (this.structtype == "list") {
  96                     this.client.rpush(k,msg.payload);
  97                 }
  98             } else {
  99                 this.warn("No key or topic set");
 100             }
 101         });
 102         this.on("close", function() {
 103             redisConnectionPool.close(node.client);
 104         });
 105     }
 106     RED.nodes.registerType("redis out",RedisOutNode);
 107 }