dgbuilder/test/red/comms_spec.js

   1 /**
   2  * Copyright 2014 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 var should = require("should");
  18 var http = require('http');
  19 var express = require('express');
  20 var app = express();
  21 var WebSocket = require('ws');
  22
  23 var comms = require("../../red/comms.js");
  24 var address = '127.0.0.1';
  25 var listenPort = 0; // use ephemeral port
  26
  27 describe("comms", function() {
  28     describe("with default keepalive", function() {
  29         var server;
  30         var url;
  31         var port;
  32         before(function(done) {
  33             server = http.createServer(function(req,res){app(req,res)});
  34             comms.init(server, {});
  35             server.listen(listenPort, address);
  36             server.on('listening', function() {
  37                 port = server.address().port;
  38                 url = 'http://' + address + ':' + port + '/comms';
  39                 comms.start();
  40                 done();
  41             });
  42         });
  43
  44         after(function() {
  45             comms.stop();
  46         });
  47
  48         it('accepts connection', function(done) {
  49             var ws = new WebSocket(url);
  50             ws.on('open', function() {
  51                 ws.close();
  52                 done();
  53             });
  54         });
  55
  56         it('publishes message after subscription', function(done) {
  57             var ws = new WebSocket(url);
  58             ws.on('open', function() {
  59                 ws.send('{"subscribe":"topic1"}');
  60                 comms.publish('topic1', 'foo');
  61             });
  62             ws.on('message', function(msg) {
  63                 msg.should.equal('{"topic":"topic1","data":"foo"}');
  64                 ws.close();
  65                 done();
  66             });
  67         });
  68
  69         it('publishes retained message for subscription', function(done) {
  70             comms.publish('topic2', 'bar', true);
  71             var ws = new WebSocket(url);
  72             ws.on('open', function() {
  73                 ws.send('{"subscribe":"topic2"}');
  74             });
  75             ws.on('message', function(msg) {
  76                 msg.should.equal('{"topic":"topic2","data":"bar"}');
  77                 ws.close();
  78                 done();
  79             });
  80         });
  81
  82         it('retained message is deleted by non-retained message', function(done) {
  83             comms.publish('topic3', 'retained', true);
  84             comms.publish('topic3', 'non-retained');
  85             var ws = new WebSocket(url);
  86             ws.on('open', function() {
  87                 ws.send('{"subscribe":"topic3"}');
  88                 comms.publish('topic3', 'new');
  89             });
  90             ws.on('message', function(msg) {
  91                 msg.should.equal('{"topic":"topic3","data":"new"}');
  92                 ws.close();
  93                 done();
  94             });
  95         });
  96
  97         it('malformed messages are ignored',function(done) {
  98             var ws = new WebSocket(url);
  99             ws.on('open', function() {
 100                 ws.send('not json');
 101                 ws.send('[]');
 102                 ws.send('{"subscribe":"topic3"}');
 103                 comms.publish('topic3', 'correct');
 104             });
 105             ws.on('message', function(msg) {
 106                 msg.should.equal('{"topic":"topic3","data":"correct"}');
 107                 ws.close();
 108                 done();
 109             });
 110         });
 111
 112         // The following test currently fails due to minimum viable
 113         // implementation. More test should be written to test topic
 114         // matching once this one is passing
 115
 116         if (0) {
 117             it('receives message on correct topic', function(done) {
 118                 var ws = new WebSocket(url);
 119                 ws.on('open', function() {
 120                     ws.send('{"subscribe":"topic4"}');
 121                     comms.publish('topic5', 'foo');
 122                     comms.publish('topic4', 'bar');
 123                 });
 124                 ws.on('message', function(msg) {
 125                     msg.should.equal('{"topic":"topic4","data":"bar"}');
 126                     ws.close();
 127                     done();
 128                 });
 129             });
 130         }
 131     });
 132
 133     describe("keep alives", function() {
 134         var server;
 135         var url;
 136         var port;
 137         before(function(done) {
 138             server = http.createServer(function(req,res){app(req,res)});
 139             comms.init(server, {webSocketKeepAliveTime: 100});
 140             server.listen(listenPort, address);
 141             server.on('listening', function() {
 142                 port = server.address().port;
 143                 url = 'http://' + address + ':' + port + '/comms';
 144                 comms.start();
 145                 done();
 146             });
 147         });
 148         after(function() {
 149             comms.stop();
 150         });
 151         it('are sent', function(done) {
 152             var ws = new WebSocket(url);
 153             var count = 0;
 154             ws.on('message', function(data) {
 155                 var msg = JSON.parse(data);
 156                 msg.should.have.property('topic','hb');
 157                 msg.should.have.property('data').be.a.Number;
 158                 count++;
 159                 if (count == 3) {
 160                     ws.close();
 161                     done();
 162                 }
 163             });
 164         });
 165         it('are not sent if other messages are sent', function(done) {
 166             var ws = new WebSocket(url);
 167             var count = 0;
 168             var interval;
 169             ws.on('open', function() {
 170                 ws.send('{"subscribe":"foo"}');
 171                 interval = setInterval(function() {
 172                     comms.publish('foo', 'bar');
 173                 }, 50);
 174             });
 175             ws.on('message', function(data) {
 176                 var msg = JSON.parse(data);
 177                 msg.should.have.property('topic', 'foo');
 178                 msg.should.have.property('data', 'bar');
 179                 count++;
 180                 if (count == 5) {
 181                     clearInterval(interval);
 182                     ws.close();
 183                     done();
 184                 }
 185             });
 186         });
 187     });
 188
 189 });