Merge pull request #61 from dmoranj/bug/proxyIntercepObserveRequests

Fix Observes through proxy

lib/helpers.js

@@ -198,3 +198,10 @@ module.exports.createBlock2 = function createBlock2(requestedBlock) {
   }
   return (extraNum)? Buffer.concat([extraNum, new Buffer([byte])]):new Buffer([byte])
 }
+
+/**
+ * Provide a or function to use with the reduce() Array method
+ */
+module.exports.or = function or(previous, current) {
+    return previous || current;
+}

lib/middlewares.js

@@ -12,6 +12,7 @@ var dgram           = require('dgram')
   , crypto          = require('crypto')
   , events          = require('events')
   , LRU             = require('lru-cache')
+  , pktToMsg        = require('./helpers').packetToMessage
   , parse           = require('coap-packet').parse
   , generate        = require('coap-packet').generate
   , IncomingMessage = require('./incoming_message')
@@ -22,6 +23,7 @@ var dgram           = require('dgram')
   , parseBlock2     = require('./helpers').parseBlock2
   , createBlock2    = require('./helpers').createBlock2
   , getOption       = require('./helpers').getOption
+  , or              = require('./helpers').or
 
 function parseRequest(request, next) {
   try {
@@ -63,6 +65,10 @@ function proxyRequest(request, next) {
   }
 }
 
+function isObserve(packet) {
+  request.packet.options.reduce(or, false);
+}
+
 function handleProxyResponse(request, next) {
   if (request.proxy) {
     return next(null)
@@ -71,13 +77,17 @@ function handleProxyResponse(request, next) {
   var originalProxiedRequest = request.server._proxiedRequests[request.packet.token.toString('hex')]
   if ( originalProxiedRequest ) {
     request.server._sendReverseProxied(request.packet, originalProxiedRequest.rsinfo)
-    delete request.server._proxiedRequests[request.packet.token.toString('hex')]
+
+    if (!isObserve(request.packet))
+      delete request.server._proxiedRequests[request.packet.token.toString('hex')]
+
     next(null)
   } else {
     next()
   }
 }
 
+
 exports.parseRequest = parseRequest
 exports.handleServerRequest = handleServerRequest
 exports.proxyRequest = proxyRequest

test/proxy.js

@@ -55,6 +55,29 @@ describe('proxy', function() {
     client.send(message, 0, message.length, port, '127.0.0.1')
   }
 
+  function ssend(rsinfo, packet) {
+    var toSend = generate(packet)
+    target.send(toSend, 0, toSend.length, rsinfo.port, rsinfo.address)
+  }
+
+  function sendObservation(message) {
+    target.on('request', function(req, res) {
+      res.setOption('Observe', 1);
+      res.write('Pruebas');
+
+      setTimeout(function() {
+        res.write('Pruebas2');
+        res.end('Last msg');
+      }, 500)
+    })
+
+    return request({
+      port: port
+      , observe: true
+      ,  proxyUri: 'coap://localhost:' + targetPort + '/the/path'
+    }).end()
+  }
+
   function fastForward(increase, max) {
     clock.tick(increase)
     if (increase < max)
@@ -74,6 +97,24 @@ describe('proxy', function() {
     })
   })
 
+  it('should resend notifications in an observe connection', function(done) {
+    var counter = 0,
+        req;
+
+    req = sendObservation()
+
+    req.on('response', function(res) {
+      res.on('data', function(msg) {
+        if (counter === 2)
+          done()
+        else
+          counter++;
+
+        clock.tick(600);
+      })
+    })
+  })
+
   it('should not process the request as a standard server request', function(done) {
     target.on('request', function(req, res) {
       done()