Merge pull request #11 from jprjr/new-preview-window

New preview window

.gitignore

@@ -3,3 +3,4 @@
 /static/favicon.ico
 /.lua-version
 *.swp
+/local/

README.md

@@ -34,6 +34,7 @@ I also have a YouTube video about installing this available here: https://youtu.
   + [Alternative: Install nginx with Lua and rtmp](#alternative-install-nginx-with-lua-and-rtmp)
   + [Setup database and user in Postgres](#setup-database-and-user-in-postgres)
   + [Setup Redis](#setup-redis)
+  + [Setup Sockexec](#setup-sockexec)
   + [Setup Authentication Server](#setup-authentication-server)
   + [Clone and setup](#clone-and-setup)
   + [Install Lua modules](#install-lua-modules)
@@ -150,6 +151,16 @@ postgres=# \q
 I'm not going to write up instructions for setting up Redis - this is more
 of a checklist item.
 
+### Setup Sockexec
+
+`multistreamer` uses the `lua-resty-exec` module for managing ffmpeg processes,
+which requires a running instance of [`sockexec`](https://github.com/jprjr/sockexec).
+The `sockexec` repo has instructions for installation - you can either compile from
+source, or just download a static binary.
+
+Make sure you change `sockexec`'s default timeout value. The default is pretty
+conservative (60 seconds). I'd recommend making it infinite (ie, `sockexec -t0 /tmp/exec.sock`).
+
 ### Setup Authentication Server
 
 `multistreamer` doesn't handle its own authentication - instead, it will
@@ -222,6 +233,7 @@ Each module has more details in the [wiki.](https://github.com/jprjr/multistream
 
 You'll need some Lua modules installed:
 
+* lua-resty-exec
 * lua-resty-jit-uuid
 * lua-resty-string
 * lua-resty-http
@@ -315,14 +327,20 @@ It's important to note that updating the web interface does *not* immediately
 change anything on the user's streaming services - it's saved for later,
 when the user starts pushing video.
 
-When the user starts pushing video to the RTMP endpoint, `multistreamer` will
-update each account as needed - like updating the Twitch's broadcast title
-and game, or make a new Live Video for Facebook. It will then start relaying
-video to the different accounts.
+The user can setup a stream to either start pushing video to their streaming
+services as soon as an incoming video stream is detected, or to wait until
+they've had a chance to preview the stream. Either way, `multistreamer` will
+update each account as needed just before it starts pushing video out - things
+like updating the Twitch's broadcast title and game, or make a new Live Video
+for Facebook.
 
 Once the user stops pushing video, `multistreamer` will take any needed
 shutdown/stop actions - like ending the Facebook Live Video.
 
+I highly recommend that users browse each network's section within the
+[Wiki](https://github.com/jprjr/multistreamer/wiki) - I tried to detail
+each of the metadata settings and what they mean/do.
+
 ### IRC Usage
 
 Users can connect to Multistreamer with an IRC client, and view their

app.lua

@@ -29,6 +29,7 @@ local len = string.len
 local insert = table.insert
 local sort = table.sort
 local streams_dict = ngx.shared.streams
+local capture = ngx.location.capture
 
 local pid = ngx.worker.pid()
 
@@ -178,6 +179,17 @@ app:match('metadata-edit', config.http_prefix .. '/metadata/:id', respond_to({
     end
     self.chat_level = self.stream:check_chat(self.user)
     self.metadata_level = self.stream:check_meta(self.user)
+
+    self.stream_status = streams_dict:get(self.stream.id)
+    if self.stream_status then
+      self.stream_status = from_json(self.stream_status)
+    else
+      self.stream_status = {
+          data_incoming = false,
+          data_pushing = false,
+          data_pulling = false,
+      }
+    end
     if self.metadata_level == 0 then
       return err_out(self,'Stream not found')
     end
@@ -200,12 +212,35 @@ app:match('metadata-edit', config.http_prefix .. '/metadata/:id', respond_to({
     if self.metadata_level < 2 then
       return err_out(self,'Nice try buddy')
     end
+
+    if self.params['customPullBtn'] ~= nil then
+      self.stream_status.data_pulling = true
+      streams_dict:set(self.stream.id, to_json(self.stream_status))
+      publish('process:start:pull', {
+        worker = pid,
+        id = self.stream.id,
+      })
+      self.session.status_msg = { type = 'success', msg = 'Custom Puller Started' }
+      return { redirect_to = self:url_for('metadata-edit') .. self.stream.id }
+    end
+
+    if self.params['customPullBtnStop'] ~= nil then
+      self.stream_status.data_pulling = false
+      streams_dict:set(self.stream.id, to_json(self.stream_status))
+      publish('process:end:pull', {
+        id = self.stream.id,
+      })
+      self.session.status_msg = { type = 'success', msg = 'Custom Puller Stopped' }
+      return { redirect_to = self:url_for('metadata-edit') .. self.stream.id }
+    end
+
     if self.params.title then
       self.stream:set('title',self.params.title)
     end
     if self.params.description then
       self.stream:set('description',self.params.description)
     end
+
     for _,account in pairs(self.accounts) do
       local ffmpeg_args = self.params['ffmpeg_args' .. '.' .. account.id]
       if ffmpeg_args and len(ffmpeg_args) > 0 then
@@ -227,9 +262,39 @@ app:match('metadata-edit', config.http_prefix .. '/metadata/:id', respond_to({
         end
       end
     end
+
     publish('stream:update',self.stream)
 
-    self.session.status_msg = { type = 'success', msg = 'Settings saved' }
+    local success_msg = 'Settings saved'
+
+    -- is there incoming data, and the user clicked the golivebutton? start the stream
+    if self.stream_status.data_incoming == true and self.stream_status.data_pushing == false and self.params['goLiveBtn'] ~= nil then
+      success_msg = 'Stream started'
+      self.stream_status.data_pushing = true
+
+      for _,account in pairs(self.accounts) do
+        local sa = StreamAccount:find({stream_id = self.stream.id, account_id = account.id})
+        local rtmp_url, err = account.network.publish_start(account:get_keystore(),sa:get_keystore())
+        if (not rtmp_url) or err then
+          return err_out(self,err)
+        end
+        sa:update({rtmp_url = rtmp_url})
+      end
+
+      publish('process:start:push', {
+        worker = pid,
+        id = self.stream.id,
+      })
+
+      publish('stream:start', {
+        worker = pid,
+        id = self.stream.id,
+        status = self.stream_status,
+      })
+    end
+
+    streams_dict:set(self.stream.id, to_json(self.stream_status))
+    self.session.status_msg = { type = 'success', msg = success_msg }
     return { redirect_to = self:url_for('metadata-edit') .. self.stream.id }
   end,
 }))
@@ -244,21 +309,45 @@ app:match('publish-start',config.http_prefix .. '/on-publish', respond_to({
       return plain_err_out(self,err)
     end
 
-    for _,v in pairs(sas) do
-      local account = v[1]
-      local sa = v[2]
-      local rtmp_url, err = account.network.publish_start(account:get_keystore(),sa:get_keystore())
-      if (not rtmp_url) or err then
-        return plain_err_out(self,err)
+    local stream_status = streams_dict:get(stream.id)
+    if stream_status then
+      stream_status = from_json(stream_status)
+    else
+      stream_status = {
+        data_incoming = false,
+        data_pushing = false,
+        data_pulling = false,
+      }
+    end
+
+    stream_status.data_incoming = true
+    streams_dict:set(stream.id, to_json(stream_status))
+
+    if stream.preview_required == 0 then
+      for _,v in pairs(sas) do
+        local account = v[1]
+        local sa = v[2]
+        local rtmp_url, err = account.network.publish_start(account:get_keystore(),sa:get_keystore())
+        if (not rtmp_url) or err then
+          return plain_err_out(self,err)
+        end
+        sa:update({rtmp_url = rtmp_url})
       end
-      sa:update({rtmp_url = rtmp_url})
+
+      stream_status.data_pushing = true
+      streams_dict:set(stream.id, to_json(stream_status))
+
+      publish('process:start:push', {
+        worker = pid,
+        id = stream.id,
+      })
+
     end
 
-    -- just going to ignore any errors
-    local ok, err = streams_dict:set(stream.id,true)
     publish('stream:start', {
       worker = pid,
       id = stream.id,
+      status = stream_status,
     })
 
     return plain_err_out(self,'OK',200)
@@ -298,9 +387,14 @@ app:post('publish-stop',config.http_prefix .. '/on-done',function(self)
     return plain_err_out(self,err)
   end
 
+  publish('process:end:push', {
+    id = stream.id,
+  })
+
   publish('stream:end', {
     id = stream.id,
   })
+
   streams_dict:set(stream.id,nil)
 
   for _,v in pairs(sas) do
@@ -353,8 +447,13 @@ app:get('site-root', config.http_prefix .. '/', function(self)
   end
 
   for k,v in pairs(self.streams) do
-    local ok = streams_dict:get(v.id)
-    if ok then
+    local stream_status = streams_dict:get(v.id)
+    if stream_status then
+      stream_status = from_json(stream_status)
+    else
+      stream_status = { data_pushing = false }
+    end
+    if stream_status.data_pushing == true then
       v.live = true
     else
       v.live = false
@@ -441,6 +540,38 @@ app:match('stream-chat', config.http_prefix .. '/stream/:id/chat', respond_to({
   end,
 }))
 
+app:match('stream-video', config.http_prefix .. '/stream/:id/video(/*)', respond_to({
+  before = function(self)
+    local stream = Stream:find({ id = self.params.id })
+    if not stream then
+      return plain_err_out(self, 'Stream not found')
+    end
+    local ok = streams_dict:get(stream.id)
+    if ok == nil or ok == 0 then
+      return plain_err_out(self, 'Stream not live', 404)
+    end
+    self.stream = stream
+  end,
+  GET = function(self)
+    local fn = self.params.splat
+    if not fn then
+      fn = 'index.m3u8'
+    end
+    local res = capture(config.http_prefix .. '/video_raw/' .. self.stream.uuid .. '/' .. fn)
+    if res then
+      if res.status == 302 then
+        return plain_err_out(self, 'Stream not live', 404)
+      end
+      return self:write({
+        layout = 'plain',
+        content_type = res.header['content-type'],
+        status = res.status,
+      }, res.body)
+    end
+    return plain_err_out(self,'An error occured', 500)
+  end,
+}))
+
 app:match('stream-ws', config.http_prefix .. '/ws/:id',respond_to({
   before = function(self)
     if not require_login(self) then

bin/multistreamer.lua

@@ -16,13 +16,15 @@ local insert = table.insert
 
 local script_path = posix.realpath(arg[0])
 local streamer_dir = posix.dirname(posix.dirname(script_path))
+local bash_path   = streamer_dir ..'/bin/multistreamer'
 posix.chdir(streamer_dir)
 
 local commands = {
   ['run'] = 1,
   ['initdb'] = 1,
   ['psql'] = 1,
   ['push'] = 1, -- internal command
+  ['pull'] = 1, -- internal command
 }
 
 local sql_files = {
@@ -31,6 +33,7 @@ local sql_files = {
   [3] = streamer_dir .. '/sql/1485029477.sql',
   [4] = streamer_dir .. '/sql/1485036089.sql',
   [5] = streamer_dir .. '/sql/1485788609.sql',
+  [6] = streamer_dir .. '/sql/1489949143.sql',
 }
 
 if(not arg[1] or not commands[arg[1]]) then
@@ -70,9 +73,14 @@ if(arg[1] == 'run') then
     print('Error: path to ffmpeg not set')
     exit(1)
   end
+  if not config.sockexec_path then
+    print('Error: path to sockexec socket not set')
+    exit(1)
+  end
 
   config.lua_bin = lua_bin
   config.script_path = script_path;
+  config.bash_path   = bash_path;
 
   if not config.work_dir then
     config.work_dir = getenv('HOME') .. '/.multistreamer'
@@ -183,6 +191,32 @@ elseif(arg[1] == 'push') then
   print(err)
   exit(1)
 
+elseif(arg[1] == 'pull') then
+  if not arg[2] then
+    print('pull requires uuid argument')
+    exit(1)
+  end
+
+  local shell = require'multistreamer.shell'
+  local StreamModel = require'models.stream'
+  local stream = StreamModel:find({uuid = arg[2]})
+
+  local ffmpeg_args = {
+    '-v',
+    'error',
+  }
+
+  local args = shell.parse(stream.ffmpeg_pull_args)
+  for i,v in pairs(args) do
+    insert(ffmpeg_args,v)
+  end
+  insert(ffmpeg_args,'-f')
+  insert(ffmpeg_args,'flv')
+  insert(ffmpeg_args,config.private_rtmp_url ..'/'..config.rtmp_prefix..'/'..arg[2])
+  local _, err = posix.exec(config.ffmpeg,ffmpeg_args)
+  print(err)
+  exit(1)
+
 end
 
 

config.lua.example

@@ -140,6 +140,10 @@ config({'development','production'}, {
   -- defaults to $HOME/.multistreamer if not set
   -- work_dir = '/path/to/some/folder',
 
+  -- set the path to sockexec's socket
+  -- see https://github.com/jprjr/sockexec for installation details
+  -- sockexec_path = '/tmp/exec.sock',
+
 })