diff --git a/lib/wsp-client.js b/lib/wsp-client.js new file mode 100644 index 00000000..8640d917 --- /dev/null +++ b/lib/wsp-client.js @@ -0,0 +1,460 @@ +/* eslint-env node*/ +/* + eslint-disable + require-jsdoc, + no-warning-comments, + no-underscore-dangle, + max-params, +*/ +'use strict'; + +var Gun = require('../gun'); +var WS = require('ws'); + +var Tab = {}; +Tab.on = Gun.on; +Tab.peers = (function () { + + function Peer (peers) { + if (!Peer.is(this)) { + return new Peer(peers); + } + + this.peers = peers; + } + + Peer.is = function (peer) { + return peer instanceof Peer; + }; + + function map (peer, url) { + var msg = this.msg; + var opt = this.opt || {}; + opt.out = true; + Peer.request(url, msg, null, opt); + } + + Peer.prototype.send = function (msg, opt) { + Peer.request.each(this.peers, map, { + msg: msg, + opt: opt, + }); + }; + + Peer.request = (function () { + + function request (base, body, cb, opt) { + + var obj = base.length ? { base: base } : {}; + obj.base = opt.base || base; + obj.body = opt.body || body; + obj.headers = opt.headers; + obj.url = opt.url; + obj.out = opt.out; + cb = cb || function () {}; + + if (!obj.base) { + return; + } + + request.transport(obj, cb); + } + + request.createServer = function (fn) { + request.createServer.list.push(fn); + }; + + request.createServer.ing = function (req, cb) { + var index = request.createServer.list.length; + var server; + while (index) { + index -= 1; + server = request.createServer.list[index] || function () {}; + server(req, cb); + } + }; + + request.createServer.list = []; + request.back = 2; + request.backoff = 2; + + request.transport = function (opt, cb) { + if (request.ws(opt, cb)) { + return; + } + }; + + request.ws = function (opt, cb, req) { + var ws; + if (!WS) { + return false; + } + + ws = request.ws.peers[opt.base]; + if (ws) { + req = req || {}; + if (opt.headers) { + req.headers = opt.headers; + } + if (opt.body) { + req.body = opt.body; + } + + if (opt.url) { + req.url = opt.url; + } + + req.headers = req.headers || {}; + + if (!opt.out && !ws.cbs[req.headers['ws-rid']]) { + var rid = 'WS' + + new Date().getTime() + + '.' + + Math.floor((Math.random() * 65535) + 1); + + req.headers['ws-rid'] = rid; + + ws.cbs[rid] = function (err, res) { + if (!res || res.body || res.end) { + delete ws.cbs[req.headers['ws-rid']]; + } + + cb(err, res); + }; + } + + if (!ws.readyState) { + setTimeout(function () { + request.ws(opt, cb, req); + }, 100); + + return true; + } + + ws.sending = true; + ws.send(JSON.stringify(req)); + return true; + } + + if (ws === false) { + return false; + } + + var wsURL = opt.base.replace('http', 'ws'); + + ws = request.ws.peers[opt.base] = new WS(wsURL); + ws.cbs = {}; + + ws.onopen = function () { + request.back = 2; + request.ws(opt, cb); + }; + + ws.onclose = function (event) { + + if (!ws || !event) { + return; + } + + if (ws.close instanceof Function) { + ws.close(); + } + + if (!ws.sending) { + ws = request.ws.peers[opt.base] = false; + request.transport(opt, cb); + return; + } + + request.each(ws.cbs, function (cb) { + cb({ + err: 'WebSocket disconnected!', + code: ws.sending ? (ws || {}).err || event.code : -1, + }); + }); + + // This will make the next request try to reconnect + ws = request.ws.peers[opt.base] = null; + + // TODO: Have the driver handle this! + setTimeout(function () { + + // opt here is a race condition, + // is it not? Does this matter? + request.ws(opt, function () {}); + }, request.back *= request.backoff); + }; + + ws.onmessage = function (msg) { + var res; + if (!msg || !msg.data) { + return; + } + try { + res = JSON.parse(msg.data); + } catch (error) { + return; + } + if (!res) { + return; + } + res.headers = res.headers || {}; + if (res.headers['ws-rid']) { + var cb = ws.cbs[res.headers['ws-rid']] || function () {}; + cb(null, res); + return; + } + + // emit extra events. + if (res.body) { + request.createServer.ing(res, function (res) { + res.out = true; + request(opt.base, null, null, res); + }); + } + }; + + ws.onerror = function (error) { + (ws || {}).err = error; + }; + + return true; + }; + request.ws.peers = {}; + request.ws.cbs = {}; + + request.each = function (obj, cb, as) { + if (!obj || !cb) { + return; + } + + for (var key in obj) { + if (obj.hasOwnProperty(key)) { + cb.call(as, obj[key], key); + } + } + }; + + return request; + }()); + + return Peer; +}()); + +// Handle read requests. +Gun.on('get', function (at) { + var gun = at.gun; + var opt = at.opt || {}; + var peers = opt.peers || gun.Back('opt.peers'); + + if (!peers || Gun.obj.empty(peers)) { + Gun.log.once('peers', 'Warning! You have no peers to connect to!'); + at.gun.Back(-1).on('in', {'@': at['#']}); + + return; + } + + // Create a new message. + var msg = { + + // msg ID + '#': at['#'] || Gun.text.random(9), + + // msg BODY + '$': at.get, + }; + + // Listen for a response. + // TODO: ONE? PERF! Clear out listeners, maybe with setTimeout? + Tab.on(msg['#'], function (err, data) { + var obj = { + '@': at['#'], + err: err, + put: data, + }; + + if (data) { + at.gun.Back(-1).on('out', obj); + } else { + at.gun.Back(-1).on('in', obj); + } + }); + + // Broadcast to all other peers. + Tab.peers(peers).send(msg, { + headers: { + 'gun-sid': Tab.server.sid, + }, + }); +}); + +// Handle write requests. +Gun.on('put', function (at) { + if (at['@']) { + return; + } + var opt = at.gun.Back('opt') || {}, peers = opt.peers; + if (!peers || Gun.obj.empty(peers)) { + Gun.log.once('peers', 'Warning! You have no peers to save to!'); + at.gun.Back(-1).on('in', {'@': at['#']}); + return; + } + if (opt.websocket === false || (at.opt && at.opt.websocket === false)) { + return; + } + var msg = { + + // msg ID + '#': at['#'] || Gun.text.random(9), + + // msg BODY + '$': at.put, + }; + + // TODO: ONE? PERF! Clear out listeners, maybe with setTimeout? + Tab.on(msg['#'], function (err, ok) { + at.gun.Back(-1).on('in', { + '@': at['#'], + err: err, + ok: ok, + }); + }); + + Tab.peers(peers).send(msg, { + headers: { + 'gun-sid': Tab.server.sid, + }, + }); +}); + +// REVIEW: Do I need this on a server client? +// browser/client side Server! +// TODO: BUG! Does not respect separate instances!!! +Gun.on('opt', function (at) { + if (Tab.server) { + return; + } + + var gun = at.gun; + var server = Tab.server = Tab.server || {}; + var tmp; + + server.sid = Gun.text.random(); + + Tab.peers.request.createServer(function (req, res) { + + // Validate request. + if (!req || !res || !req.body || !req.headers) { + return; + } + + var msg = req.body; + + // AUTH for non-replies. + if (server.msg(msg['#'])) { + return; + } + + // no need to process. + if (msg['@']) { + if (Tab.ons[tmp = msg['@'] || msg['#']]) { + Tab.on(tmp, [msg['!'], msg.$]); + } + return; + } + + if (msg.$ && msg.$['#']) { + server.get(req, res); + return; + } + + server.put(req, res); + }); + + server.get = function (req, cb) { + var body = req.body; + var lex = body.$; + var graph = gun._.root._.graph; + var node; + + // Don't reply to data we don't have it in memory. + // TODO: Add localStorage? + if (!(node = graph[lex['#']])) { + return; + } + + cb({ + body: { + '#': server.msg(), + '@': body['#'], + '$': node, + }, + }); + }; + + server.put = function (req, cb) { + var body = req.body, graph = body.$; + var __ = gun._.root._; + + // filter out what we don't have in memory. + if (!(graph = Gun.obj.map(graph, function (node, soul, map) { + if (!__.path[soul]) { + return; + } + map(soul, node); + }))) { + return; + } + gun.on('out', { + gun: gun, + opt: { + websocket: false, + }, + put: graph, + '#': Gun.on.ask(function (ack, ev) { + if (!ack) { + return undefined; + } + ev.off(); + return cb({ + body: { + '#': server.msg(), + '@': body['#'], + '$': ack, + '!': ack.err, + }, + }); + }), + }); + }; + + server.msg = function (id) { + if (!id) { + id = Gun.text.random(9); + server.msg.debounce[id] = Gun.time.is(); + return id; + } + + clearTimeout(server.msg.clear); + server.msg.clear = setTimeout(function () { + var now = Gun.time.is(); + Gun.obj.map(server.msg.debounce, function (time, id) { + if ((now - time) < (1000 * 60 * 5)) { + return; + } + + Gun.obj.del(server.msg.debounce, id); + }); + }, 500); + + if (server.msg.debounce[id]) { + server.msg.debounce[id] = Gun.time.is(); + return id; + } + + server.msg.debounce[id] = Gun.time.is(); + return undefined; + }; + + server.msg.debounce = server.msg.debounce || {}; +}); diff --git a/lib/wsp.js b/lib/wsp.js index 7768fa86..7ac4bbac 100644 --- a/lib/wsp.js +++ b/lib/wsp.js @@ -1,230 +1,184 @@ -;(function(wsp){ - /* - TODO: SERVER PUSH! - TODO: SERVER GET! - TODO: SERVER PUSH! - TODO: SERVER GET! - TODO: SERVER PUSH! - TODO: SERVER GET! - TODO: SERVER PUSH! - TODO: SERVER GET! - TODO: SERVER PUSH! - TODO: SERVER GET! - TODO: SERVER PUSH! - TODO: SERVER GET! - */ - var Gun = require('../gun') - , formidable = require('formidable') - , ws = require('ws').Server - , http = require('./http') - , url = require('url'); - Gun.on('opt', function(at){ - var gun = at.gun, opt = at.opt; - gun.__ = at.root._; - gun.__.opt.ws = opt.ws = gun.__.opt.ws || opt.ws || {}; - function start(server, port, app){ - if(app && app.use){ app.use(gun.wsp.server) } - server = gun.__.opt.ws.server = gun.__.opt.ws.server || opt.ws.server || server; - require('./ws')(gun.wsp.ws = gun.wsp.ws || new ws(gun.__.opt.ws), function(req, res){ - var ws = this; - req.headers['gun-sid'] = ws.sid = ws.sid? ws.sid : req.headers['gun-sid']; - ws.sub = ws.sub || gun.wsp.on('network', function(msg, ev){ - if(!ws || !ws.send || !ws._socket || !ws._socket.writable){ return ev.off() } - if(!msg || (msg.headers && msg.headers['gun-sid'] === ws.sid)){ return } - if(msg && msg.headers){ delete msg.headers['ws-rid'] } - // TODO: BUG? ^ What if other peers want to ack? Do they use the ws-rid or a gun declared id? - try{ws.send(Gun.text.ify(msg)); - }catch(e){} // juuuust in case. - }); - gun.wsp.wire(req, res); - }, {headers: {'ws-rid': 1, 'gun-sid': 1}}); - gun.__.opt.ws.port = gun.__.opt.ws.port || opt.ws.port || port || 80; - } - var wsp = gun.wsp = gun.wsp || function(server){ - if(!server){ return gun } - if(Gun.fns.is(server.address)){ - if(server.address()){ - start(server, server.address().port); - return gun; - } - } - if(Gun.fns.is(server.get) && server.get('port')){ - start(server, server.get('port')); - return gun; - } - var listen = server.listen; - server.listen = function(port){ - var serve = listen.apply(server, arguments); - start(serve, port, server); - return serve; - } - return gun; - } - gun.wsp.on = gun.wsp.on || Gun.on; - gun.wsp.regex = gun.wsp.regex || opt.route || opt.path || /^\/gun/i; - gun.wsp.poll = gun.wsp.poll || opt.poll || 1; - gun.wsp.pull = gun.wsp.pull || opt.pull || gun.wsp.poll * 1000; - gun.wsp.server = gun.wsp.server || function(req, res, next){ // http - next = next || function(){}; - if(!req || !res){ return next(), false } - if(!req.url){ return next(), false } - if(!req.method){ return next(), false } - var msg = {}; - msg.url = url.parse(req.url, true); - if(!gun.wsp.regex.test(msg.url.pathname)){ return next(), false } // TODO: BUG! If the option isn't a regex then this will fail! - if(msg.url.pathname.replace(gun.wsp.regex,'').slice(0,3).toLowerCase() === '.js'){ - res.writeHead(200, {'Content-Type': 'text/javascript'}); - res.end(gun.wsp.js = gun.wsp.js || require('fs').readFileSync(__dirname + '/../gun.js')); // gun server is caching the gun library for the client - return true; +/* eslint-disable*/ +var Gun = require('../gun') +, formidable = require('formidable') +, http = require('./http') +, url = require('url') +, wsp = {} +, WS = require('ws') +, ws = WS.Server; + +// Handles server to server sync. +require('./wsp-client.js'); + +Gun.on('opt', function(at){ + var gun = at.gun, opt = at.opt; + gun.__ = at.root._; + gun.__.opt.ws = opt.ws = gun.__.opt.ws || opt.ws || {}; + + function start(server, port, app){ + if(app && app.use){ app.use(gun.wsp.server) } + server = gun.__.opt.ws.server = gun.__.opt.ws.server || opt.ws.server || server; + require('./ws')(gun.wsp.ws = gun.wsp.ws || new ws(gun.__.opt.ws), function(req, res){ + var ws = this; + req.headers['gun-sid'] = ws.sid = ws.sid? ws.sid : req.headers['gun-sid']; + ws.sub = ws.sub || gun.wsp.on('network', function(msg, ev){ + if(!ws || !ws.send || !ws._socket || !ws._socket.writable){ return ev.off() } + if(!msg || (msg.headers && msg.headers['gun-sid'] === ws.sid)){ return } + if(msg && msg.headers){ delete msg.headers['ws-rid'] } + // TODO: BUG? ^ What if other peers want to ack? Do they use the ws-rid or a gun declared id? + try{ws.send(Gun.text.ify(msg)); + }catch(e){} // juuuust in case. + }); + gun.wsp.wire(req, res); + }, {headers: {'ws-rid': 1, 'gun-sid': 1}}); + gun.__.opt.ws.port = gun.__.opt.ws.port || opt.ws.port || port || 80; + } + var wsp = gun.wsp = gun.wsp || function(server){ + if(!server){ return gun } + if(Gun.fns.is(server.address)){ + if(server.address()){ + start(server, server.address().port); + return gun; } - if(!req.upgrade){ return next(), false } - return http(req, res, function(req, res){ - if(!req){ return next() } - var stream, cb = res = require('./jsonp')(req, res); - if(req.headers && (stream = req.headers['gun-sid'])){ - stream = (gun.wsp.peers = gun.wsp.peers || {})[stream] = gun.wsp.peers[stream] || {sid: stream}; - stream.drain = stream.drain || function(res){ - if(!res || !stream || !stream.queue || !stream.queue.length){ return } - res({headers: {'gun-sid': stream.sid}, body: stream.queue }); - stream.off = setTimeout(function(){ stream = null }, gun.wsp.pull); - stream.reply = stream.queue = null; - return true; - } - stream.sub = stream.sub || gun.wsp.on('network', function(req, ev){ - if(!stream){ return ev.off() } // self cleans up after itself! - if(!req || (req.headers && req.headers['gun-sid'] === stream.sid)){ return } - (stream.queue = stream.queue || []).push(req); - stream.drain(stream.reply); - }); - cb = function(r){ (r.headers||{}).poll = gun.wsp.poll; res(r) } - clearTimeout(stream.off); - if(req.headers.pull){ - if(stream.drain(cb)){ return } - return stream.reply = cb; - } - } - gun.wsp.wire(req, cb); - }), true; - } - if((gun.__.opt.maxSockets = opt.maxSockets || gun.__.opt.maxSockets) !== false){ - require('https').globalAgent.maxSockets = require('http').globalAgent.maxSockets = gun.__.opt.maxSockets || Infinity; - } - gun.wsp.msg = gun.wsp.msg || function(id){ - if(!id){ - return gun.wsp.msg.debounce[id = Gun.text.random(9)] = Gun.time.is(), id; - } - clearTimeout(gun.wsp.msg.clear); - gun.wsp.msg.clear = setTimeout(function(){ - var now = Gun.time.is(); - Gun.obj.map(gun.wsp.msg.debounce, function(t,id){ - if((now - t) < (1000 * 60 * 5)){ return } - Gun.obj.del(gun.wsp.msg.debounce, id); - }); - },500); - if(id = gun.wsp.msg.debounce[id]){ - return gun.wsp.msg.debounce[id] = Gun.time.is(), id; - } - gun.wsp.msg.debounce[id] = Gun.time.is(); - return; - }; - gun.wsp.msg.debounce = gun.wsp.msg.debounce || {}; - gun.wsp.wire = gun.wsp.wire || (function(){ - // all streams, technically PATCH but implemented as PUT or POST, are forwarded to other trusted peers - // except for the ones that are listed in the message as having already been sending to. - // all states, implemented with GET, are replied to the source that asked for it. - function tran(req, res){ - if(!req || !res || !req.body || !req.headers){ return } - if(req.url){ req.url = url.format(req.url) } - var msg = req.body; - // AUTH for non-replies. - if(gun.wsp.msg(msg['#'])){ return } - gun.wsp.on('network', Gun.obj.copy(req)); - if(msg['@']){ return } // no need to process. - if(msg['$'] && msg['$']['#']){ return tran.get(req, res) } - //if(Gun.is.lex(msg['$'])){ return tran.get(req, res) } - else { return tran.put(req, res) } - cb({body: {hello: 'world'}}); - // TODO: BUG! server put should push. - } - tran.get = function(req, cb){ - var body = req.body, lex = body['$'], reply = {headers: {'Content-Type': tran.json}}, opt; - gun.on('out', {gun: gun, get: lex, req: 1, '#': Gun.on.ask(function(at, ev){ - ev.off(); - var graph = at.put; - return cb({headers: reply.headers, body: { - '#': gun.wsp.msg(), - '@': body['#'], - '$': graph, - '!': at.err - }}); - return; - if(Gun.obj.empty(node)){ - return cb({headers: reply.headers, body: node}); - } // we're out of stuff! - /* - (function(chunks){ // FEATURE! Stream chunks if the nodes are large! - var max = 10, count = 0, soul = Gun.is.node.soul(node); - if(Object.keys(node).length > max){ - var n = Gun.is.node.soul.ify({}, soul); - Gun.obj.map(node, function(val, field){ - if(!(++count % max)){ - cb({headers: reply.headers, chunk: n}); // send node chunks - n = Gun.is.node.soul.ify({}, soul); - } - Gun.is.node.state.ify([n, node], field, val); - }); - if(count % max){ // finish off the last chunk - cb({headers: reply.headers, chunk: n}); - } - } else { - cb({headers: reply.headers, chunk: node}); // send full node - } - }([])); - */ - cb({headers: reply.headers, chunk: node }); // Use this if you don't want streaming chunks feature. - })}); - } - tran.put = function(req, cb){ - //console.log("tran.put", req); - // NOTE: It is highly recommended you do your own PUT/POSTs through your own API that then saves to gun manually. - // This will give you much more fine-grain control over security, transactions, and what not. - var body = req.body, graph = body['$'], reply = {headers: {'Content-Type': tran.json}}, opt; - gun.on('out', {gun: gun, put: graph, '#': Gun.on.ask(function(ack, ev){ - //Gun.on('put', {gun: gun, put: graph, '#': Gun.on.ask(function(ack, ev){ - ev.off(); - return cb({headers: reply.headers, body: { - '#': gun.wsp.msg(), - '@': body['#'], - '$': ack, - '!': ack.err - }}); - })}); - return; - if(Gun.is.graph(req.body)){ - if(req.err = Gun.union(gun, req.body, function(err, ctx){ // TODO: BUG? Probably should give me ctx.graph - if(err){ return cb({headers: reply.headers, body: {err: err || "Union failed."}}) } - var ctx = ctx || {}; ctx.graph = {}; - Gun.is.graph(req.body, function(node, soul){ - ctx.graph[soul] = gun.__.graph[soul]; - }); - (gun.__.opt.wire.put || function(g,cb){cb("No save.")})(ctx.graph, function(err, ok){ - if(err){ return cb({headers: reply.headers, body: {err: err || "Failed."}}) } // TODO: err should already be an error object? - cb({headers: reply.headers, body: {ok: ok || "Persisted."}}); - //console.log("tran.put <------------------------", ok); - }); - }).err){ cb({headers: reply.headers, body: {err: req.err || "Union failed."}}) } - } else { - cb({headers: reply.headers, body: {err: "Not a valid graph!"}}); - } - } - gun.wsp.on('network', function(req){ - // TODO: MARK! You should move the networking events to here, not in WSS only. - }); - tran.json = 'application/json'; - return tran; - }()); - if(opt.server){ - wsp(opt.server); - } - }); -}({})); + } + if(Gun.fns.is(server.get) && server.get('port')){ + start(server, server.get('port')); + return gun; + } + var listen = server.listen; + server.listen = function(port){ + var serve = listen.apply(server, arguments); + start(serve, port, server); + return serve; + } + return gun; + } + gun.wsp.on = gun.wsp.on || Gun.on; + gun.wsp.regex = gun.wsp.regex || opt.route || opt.path || /^\/gun/i; + gun.wsp.poll = gun.wsp.poll || opt.poll || 1; + gun.wsp.pull = gun.wsp.pull || opt.pull || gun.wsp.poll * 1000; + gun.wsp.server = gun.wsp.server || function(req, res, next){ // http + next = next || function(){}; + if(!req || !res){ return next(), false } + if(!req.url){ return next(), false } + if(!req.method){ return next(), false } + var msg = {}; + msg.url = url.parse(req.url, true); + if(!gun.wsp.regex.test(msg.url.pathname)){ return next(), false } // TODO: BUG! If the option isn't a regex then this will fail! + if(msg.url.pathname.replace(gun.wsp.regex,'').slice(0,3).toLowerCase() === '.js'){ + res.writeHead(200, {'Content-Type': 'text/javascript'}); + res.end(gun.wsp.js = gun.wsp.js || require('fs').readFileSync(__dirname + '/../gun.js')); // gun server is caching the gun library for the client + return true; + } + + if(!req.upgrade){ + next(); + return false; + } + + return http(req, res, function(req, res){ + if(!req){ return next() } + var stream, cb = res = require('./jsonp')(req, res); + if(req.headers && (stream = req.headers['gun-sid'])){ + stream = (gun.wsp.peers = gun.wsp.peers || {})[stream] = gun.wsp.peers[stream] || {sid: stream}; + stream.drain = stream.drain || function(res){ + if(!res || !stream || !stream.queue || !stream.queue.length){ return } + res({headers: {'gun-sid': stream.sid}, body: stream.queue }); + stream.off = setTimeout(function(){ stream = null }, gun.wsp.pull); + stream.reply = stream.queue = null; + return true; + } + stream.sub = stream.sub || gun.wsp.on('network', function(req, ev){ + if(!stream){ return ev.off() } // self cleans up after itself! + if(!req || (req.headers && req.headers['gun-sid'] === stream.sid)){ return } + (stream.queue = stream.queue || []).push(req); + stream.drain(stream.reply); + }); + cb = function(r){ (r.headers||{}).poll = gun.wsp.poll; res(r) } + clearTimeout(stream.off); + if(req.headers.pull){ + if(stream.drain(cb)){ return } + return stream.reply = cb; + } + } + gun.wsp.wire(req, cb); + }), true; + } + if((gun.__.opt.maxSockets = opt.maxSockets || gun.__.opt.maxSockets) !== false){ + require('https').globalAgent.maxSockets = require('http').globalAgent.maxSockets = gun.__.opt.maxSockets || Infinity; + } + gun.wsp.msg = gun.wsp.msg || function(id){ + if(!id){ + return gun.wsp.msg.debounce[id = Gun.text.random(9)] = Gun.time.is(), id; + } + clearTimeout(gun.wsp.msg.clear); + gun.wsp.msg.clear = setTimeout(function(){ + var now = Gun.time.is(); + Gun.obj.map(gun.wsp.msg.debounce, function(t,id){ + if((now - t) < (1000 * 60 * 5)){ return } + Gun.obj.del(gun.wsp.msg.debounce, id); + }); + },500); + if(id = gun.wsp.msg.debounce[id]){ + return gun.wsp.msg.debounce[id] = Gun.time.is(), id; + } + gun.wsp.msg.debounce[id] = Gun.time.is(); + return; + }; + gun.wsp.msg.debounce = gun.wsp.msg.debounce || {}; + gun.wsp.wire = gun.wsp.wire || (function(){ + // all streams, technically PATCH but implemented as PUT or POST, are forwarded to other trusted peers + // except for the ones that are listed in the message as having already been sending to. + // all states, implemented with GET, are replied to the source that asked for it. + function tran(req, res){ + if(!req || !res || !req.body || !req.headers){ return } + if(req.url){ req.url = url.format(req.url) } + var msg = req.body; + // AUTH for non-replies. + if(gun.wsp.msg(msg['#'])){ return } + gun.wsp.on('network', Gun.obj.copy(req)); + if(msg['@']){ return } // no need to process. + if(msg['$'] && msg['$']['#']){ return tran.get(req, res) } + //if(Gun.is.lex(msg['$'])){ return tran.get(req, res) } + else { return tran.put(req, res) } + cb({body: {hello: 'world'}}); + // TODO: BUG! server put should push. + } + tran.get = function(req, cb){ + var body = req.body, lex = body['$'], reply = {headers: {'Content-Type': tran.json}}, opt; + gun.on('out', {gun: gun, get: lex, req: 1, '#': Gun.on.ask(function(at, ev){ + ev.off(); + var graph = at.put; + return cb({headers: reply.headers, body: { + '#': gun.wsp.msg(), + '@': body['#'], + '$': graph, + '!': at.err + }}); + })}); + } + tran.put = function(req, cb){ + // NOTE: It is highly recommended you do your own PUT/POSTs through your own API that then saves to gun manually. + // This will give you much more fine-grain control over security, transactions, and what not. + var body = req.body, graph = body['$'], reply = {headers: {'Content-Type': tran.json}}, opt; + gun.on('out', {gun: gun, put: graph, '#': Gun.on.ask(function(ack, ev){ + //Gun.on('put', {gun: gun, put: graph, '#': Gun.on.ask(function(ack, ev){ + ev.off(); + return cb({headers: reply.headers, body: { + '#': gun.wsp.msg(), + '@': body['#'], + '$': ack, + '!': ack.err + }}); + })}); + } + gun.wsp.on('network', function(req){ + // TODO: MARK! You should move the networking events to here, not in WSS only. + }); + tran.json = 'application/json'; + return tran; + }()); + if(opt.server){ + wsp(opt.server); + } +});