litejs/app/net/index.js

var URL = require("url")
, net = require("net")
, Agent = require("http").Agent
, agentPool = {}
, debug = require("../../lib/log.js")("app:net")
, querystring = require("querystring")

, adapter = {
	http:  http,     // port 80
	https: http,     // port 443
	mqtt: na,        // port 1883
	mqtts: na,       // port 8883
	amqp: na,        // port 5672
	amqps: na,       // port 5671
	tcp: function(opts, next) {
		var socket = new net.Socket()
		, body = ""
		socket.connect(opts.port, opts.hostname, function() {
			if (opts.hParam.noDelay === "true") {
				socket.setNoDelay(true)
			}
			if (opts.body) {
				socket.write(opts.hParam.encoding ?
					Buffer.from(opts.body, opts.hParam.encoding) :
					opts.body
				)
			}
			socket.end()
		})
		socket.on("error", debug.error)
		socket.on("data", function(d) {
			body += d
		})
		socket.on("close", function() {
			if (next) {
				next(null, socket, body)
			}
		})
	},
	udp: function(opts, next) {
		var socket = require("dgram")
		.createSocket({
			reuseAddr: true,
			type: net.isIPv6(opts.hostname) ? "udp6" : "udp4"
		})
		socket.on("error", debug.error)

		socket.bind(function() {
			var message = opts.body
			, len = message.length
			if (opts.hParam.broadcast == "true") {
				socket.setBroadcast(true)
			}
			if (opts.hParam.ttl) {
				socket.setTTL(+opts.hParam.ttl)
			}
			if (opts.hParam.encoding) {
				message = Buffer.from(message, opts.hParam.encoding)
			}
			socket.send(message, opts.port, opts.hostname, function(err) {
				socket.close()
				if (next) {
					next(err, socket)
				}
			})
		})
	}
}


exports.request = request


function request(url, opts, next) {
	try {
		var req = validate(url)

		if (!next && typeof opts === "function") {
			next = opts
			opts = null
		}

		if (opts) {
			req.method = opts.method
			req.headers = opts.headers
			req.body = opts.body
		}

		debug("%s %s", (req.method || "REQ"), url)
		adapter[req.schema](req, next)
	} catch(e) {
		debug.error(e)
	}
}

function validate(url) {
	var req = typeof url === "string" ? URL.parse(url) : url
	, schema = req.schema = req.protocol.slice(0, -1)
	req.hParam = req.hash ? querystring.parse(req.hash.slice(1)) : {}
	if (!adapter[schema]) {
		throw "Invalid schema: " + req.schema
	}
	return req
}

var unsentHttp = []

function http(opts, next) {
	if (opts.hParam.keepAlive) {
		var time = +opts.hParam.keepAlive
		if (time) {
			opts.agent = agentPool[time] || (agentPool[time] = new Agent({
				keepAlive: true,
				keepAliveMsecs: time
			}))
		}
	}
	var req = require(opts.schema).request(opts, function(res) {
		if (res.statusCode < 200 || res.statusCode > 299) {
			return onError("Error: status " + res.statusCode)
		}
		var body = ""
		res.on("data", function(d) {
			body += d
		})
		res.on("end", function() {
			if (next) {
				next(null, res, body)
			}
		})
	})

	if (opts.hParam.timeout > 0) {
		req.setTimeout(+opts.hParam.timeout, function() {
			req.abort()
		})
	}
	req.on("error", onError)

	if (opts.body) {
		req.write(opts.body)
	}

	req.end()

	function onError(e) {
		debug.error("%s on %s, retry: %i", e, opts.href, opts.hParam.retry)
		if (opts.hParam.retry > 0) {
			opts.hParam.retry -= 1
			if (unsentHttp.push([opts, next]) !== 1) return
			setTimeout(function() {
				debug("Resending %i requests", unsentHttp.length)
				unsentHttp.splice(0).forEach(function(row) {
					http(row[0], row[1])
				})
			}, 60000).unref()
		}
	}
}

function na(opts, next) {
	throw opts.schema + " not implemented"
}

// PostgreSQL max query size 1 gigabyte
// SQLITE_MAX_SQL_LENGTH which defaults to 1000000
// mysql:
// SHOW VARIABLES LIKE 'max_allowed_packet';
// mysql max_allowed_packet=1048576 or 1mb


/*
Protocol	Purpose			Normal port	SSL variant	SSL port
SMTP		Send email		25/587		SMTPS		465 (legacy)[5][dead link]
POP3		Retrieve email		110		POP3S		995
IMAP		Read email		143		IMAPS		993
NNTP		News reader		119/433		NNTPS		563
LDAP		Directory Access	389		LDAPS		636
FTP		File transfer		21		FTPS		990
	// https://www.iana.org/assignments/uri-schemes/uri-schemes.xhtml
	// tag:sandro@hawke.org,2001-06-05:Taiko. http://www.taguri.org/
	// dns:[//authority/]domain[?CLASS=class;TYPE=type]
	// geo:13.4125,103.8667    lat,lon,alt https://tools.ietf.org/html/rfc5870
	// mailto:infobot@example.com?cc=bob@example.com&subject=current-issue&body=send%20current-issue
	// mqtt[s]://[username][:password]@host.domain[:port]
	// ftp:   require("./ftp"),    // port 21
	// imap://<iserver>/<enc-mailbox>[<uidvalidity>]<iuid>[<isection>][<ipartial>][<iurlauth>]
	// imap://michael@example.org/INBOX
	// imap://joe@example.com/INBOX/;uid=20/;section=1.2;urlauth=submit+fred:internal:91354a473744909de610943775f92038
	// imap:  require("./imap"),   // port 143
	// postgresql://host1:123,host2:456/somedb?target_session_attrs=any&application_name=myapp
	// smtps://address%40gmail.com:password@smtp.gmail.com
	// smtp:  require("./smtp"),   // port 25
	// ssdp:  require("./ssdp")    // port 1900
*/