Janus Lua script for the Astricon 2017 Dangerous Demo

astricon.lua

-- This is the Astricon Dangerous Demo application logic built in Lua,
-- and conceived to be used in conjunction with the janus_lua.c plugin.
-- It only uses data channels, and expects commands to control the ARI.
--
-- Note: this example depends on lua-json to do JSON processing
-- (http://luaforge.net/projects/luajson/) and on lua-http for HTTP
-- requests (https://daurnimator.github.io/lua-http/)
json = require('json')

-- Example details
name = "astricon.lua"
print("[" .. name .. "] Loading...")

-- State and properties
sessions = {}
tasks = {}
datas = {}

-- SDP template
sdpTemplate= "v=0\r\n" ..
		"o=- 0 0 IN IP4 127.0.0.1\r\n" ..
		"s=Janus TextRoom plugin\r\n" ..
		"t=0 0\r\n" ..
		"m=application 1 DTLS/SCTP 5000\r\n" ..
		"c=IN IP4 1.1.1.1\r\n" ..
		"a=sctpmap:5000 webrtc-datachannel 16\r\n"

-- ARI
ari = "http://localhost:8888/ari/channels"
ariKey = "?api_key=janus:astricon"

-- Methods
function init()
	-- This is where we initialize the plugin, for static properties
	print("[" .. name .. "] Initialized")
end

function destroy()
	-- This is where we deinitialize the plugin, when Janus shuts down
	print("[" .. name .. "] Deinitialized")
end

function createSession(id)
	-- Keep track of a new session
	print("[" .. name .. "] Created new session: " .. id)
	sessions[id] = { id = id, lua = name }
end

function destroySession(id)
	-- A Janus plugin session has gone
	print("[" .. name .. "] Destroyed session: " .. id)
	hangupMedia(id)
	sessions[id] = nil
end

function querySession(id)
	-- Return info on a session
	print("[" .. name .. "] Queried session: " .. id)
	local s = sessions[id]
	if s == nil then
		return nil
	end
	info = { script = s["lua"], id = s["id"] }
	infojson = json.encode(info)
	return infojson
end

function handleMessage(id, tr, msg, jsep)
	-- Handle a message, synchronously or asynchronously, and return
	-- something accordingly: if it's the latter, we'll do a coroutine
	print("[" .. name .. "] Handling message for session: " .. id)
	local s = sessions[id]
	if s == nil then
		return -1, "Session not found"
	end
	-- Decode the message JSON string to a table
	local msgT = json.decode(msg)
	-- We only support request:"setup" and request:"ack" here
	if msgT["request"] == "setup" then
		-- We need a new coroutine here
		async = coroutine.create(function(id, tr, comsg, cojsep)
			-- We'll only execute this when the scheduler resumes the task
			print("[" .. name .. "] Handling async message for session: " .. id)
			-- Prepare an offer
			local event = { event = "success" }
			eventjson = json.encode(event)
			local offer = { type = "offer", sdp = sdpTemplate }
			jsep = json.encode(offer)
			pushEvent(id, tr, eventjson, jsep)
		end)
		-- Enqueue it: the scheduler will resume it later
		tasks[#tasks+1] = { co = async, id = id, tr = tr, msg = msgT, jsep = nil }
		-- Return explaining that this is will be handled asynchronously
		pokeScheduler()
		return 1, nil
	elseif msgT["request"] == "ack" then
		if jsep ~= nil then
			local response = { response = "error", error = "Missing answer" }
			responsejson = json.encode(response)
			return 0, responsejson
		end
		local response = { echotest = "response", result = "success" }
		responsejson = json.encode(response)
		return 0, responsejson
	else
		local response = { response = "error", error = "Unsupported request" }
		responsejson = json.encode(response)
		return 0, responsejson
	end
end

function setupMedia(id)
	-- WebRTC is now available
	print("[" .. name .. "] WebRTC PeerConnection is up for session: " .. id)
end

function hangupMedia(id)
	-- WebRTC not available anymore
	print("[" .. name .. "] WebRTC PeerConnection is down for session: " .. id)
end

function incomingData(id, data, length)
	-- Incoming data channel message: parse and handle
	print("[" .. name .. "] Got data channel message: " .. id)
	-- Handle the data asynchronously
	async = coroutine.create(function(id, data, length)
		-- We'll only execute this when the scheduler resumes the task
		print("[" .. name .. "] Handling async data for session: " .. id)
		-- Check what we need to do
		local op = data:split(" ")
		if op[1] == "help" then
			response =
				"Welcome, brave user, to this Dangerous Demo!\n" ..
				"This is the list of supported commands:\n\n" ..
				"help                     -- Print this message\n" ..
				"channels                 -- List the active channels\n" ..
				"call USER from EXTENSION -- Originate a SIP call\n" ..
				"hangup CHANNEL           -- Hangup a channel\n" ..
				"raise hell               -- Break this demo\n"
			relayData(id, response, string.len(response))
		elseif op[1] == "channels" then
			-- Send the GET to the ARI
			local http_request = require "http.request"
			local uri = ari .. ariKey
			local post = http_request.new_from_uri(uri)
			local headers, stream = post:go()
			local body = stream:get_body_as_string()
			if body == "[]" then
				response = "No channel active"
				relayData(id, response, string.len(response))
			else
				local res = json.decode(body)
				response = "Channels:\n"
				for index,channel in ipairs(res) do
					response = response .. "  -- " .. channel["id"] .. " (" .. channel["name"] .. ", state is " .. channel["state"] .. ")\n"
				end
				relayData(id, response, string.len(response))
			end
		elseif op[1] == "call" then
			if op[2] == nil or op[3] == nil or op[3] ~= "from" or op[4] == nil then
				response = "Invalid arguments"
				relayData(id, response, string.len(response))
			end
			-- Send the POST to the ARI
			local http_request = require "http.request"
			local uri = ari .. ariKey .. "&endpoint=" .. op[2] .. "&extension=" .. op[4]
			local post = http_request.new_from_uri(uri)
			post.headers:upsert(":method", "POST")
			local headers, stream = post:go()
			local body = stream:get_body_as_string()
			local res = json.decode(body)
			local channel = res["id"]
			-- Done
			response = "Channel " .. channel
			relayData(id, response, string.len(response))
		elseif op[1] == "hangup" then
			if op[2] == nil then
				response = "Invalid arguments"
				relayData(id, response, string.len(response))
			end
			-- Send the DELETE to the ARI
			local http_request = require "http.request"
			local uri = ari .. "/" .. op[2] .. ariKey
			local post = http_request.new_from_uri(uri)
			post.headers:upsert(":method", "DELETE")
			local headers, stream = post:go()
			local body = stream:get_body_as_string()
			-- Done
			response = "Channel closed"
			relayData(id, response, string.len(response))
		elseif op[1] == "raise" and op[2] == "hell" then
			os.exit()
		else
			response = "Invalid request"
			relayData(id, response, string.len(response))
		end
	end)
	-- Enqueue it: the scheduler will resume it later
	datas[#datas+1] = { co = async, id = id, data = data, length = length }
	pokeScheduler()
end

function resumeScheduler()
	-- This is the function responsible for resuming coroutines associated
	-- with asynchronous requests: if you're handling async stuff yourself,
	-- you're free not to use this and just return, but the C Lua plugin
	-- expects this method to exist so it MUST be present, even if empty
	print("[" .. name .. "] Resuming coroutines")
	for index,task in ipairs(tasks) do
		coroutine.resume(task.co, task.id, task.tr, task.msg, task.jsep)
	end
	tasks = {}
	for index,task in ipairs(datas) do
		coroutine.resume(task.co, task.id, task.data, task.length)
	end
	print("[" .. name .. "] Coroutines resumed")
	datas = {}
end

-- Helper for logging tables
-- https://stackoverflow.com/a/27028488
function dumpTable(o)
	if type(o) == 'table' then
		local s = '{ '
		for k,v in pairs(o) do
			if type(k) ~= 'number' then k = '"'..k..'"' end
			s = s .. '['..k..'] = ' .. dumpTable(v) .. ','
		end
		return s .. '} '
	else
		return tostring(o)
	end
end

-- Helper to split strings
-- http://lua-users.org/wiki/SplitJoin
function string:split(sep)
	local sep, fields = sep or ":", {}
	local pattern = string.format("([^%s]+)", sep)
	self:gsub(pattern, function(c) fields[#fields+1] = c end)
	return fields
end


-- Done
print("[" .. name .. "]" .. " Loaded")