Merge pull request #925 from Bortlesboat/multiplayer-session-foundation

Add a reusable multiplayer session layer

src/link/LinkState.lua

@@ -10,6 +10,7 @@ local Net = require("src.link.Net")
 local Protocol = require("src.link.Protocol")
 local Runtime = require("src.mods.Runtime")
 local Screens = require("src.ui.Screens")
+local Session = require("src.link.Session")
 local TextBox = require("src.render.TextBox")
 local Strings = require("src.core.Strings")
 
@@ -44,10 +45,8 @@ local function forceLevelLabel(v)
   return (v == ANY or v == nil) and "ANY" or ("AUTO " .. tostring(v))
 end
 
--- stages before any Net object is meaningfully "this session's link" --
--- .net can still be a leftover failed attempt sitting on self, so error/
--- closed checks below skip these rather than keying off self.net's
--- presence alone
+-- stages before a successful transport has become this link's Session;
+-- terminal checks skip them rather than keying off self.net's presence
 local PRE_CONNECT_STAGES = { menu = true, lanMenu = true, onlineMenu = true }
 
 -- how long the host waits for a v2 hello before deciding the peer predates
@@ -70,6 +69,16 @@ local function ipDigits(ip)
   return digits
 end
 
+local function openSession(role, connect)
+  local transport = Net.new()
+  if connect(transport) then
+    return Session.new(transport, { role = role, kind = "link" })
+  end
+  local detail = transport.error or "?"
+  transport:close()
+  return nil, detail
+end
+
 function LinkState.new(game)
   local self = setmetatable({}, LinkState)
   self.game = game
@@ -89,12 +98,15 @@ end
 -- "connecting with this code", same as if the player had typed it in
 function LinkState.newJoinOnline(game, code)
   local self = LinkState.new(game)
-  self.net = Net.new()
-  if self.net:joinOnline(nil, code) then
+  local session, detail = openSession("guest", function(transport)
+    return transport:joinOnline(nil, code)
+  end)
+  if session then
+    self.net = session
     self.stage = "onlineJoining"
   else
     self.stage = "menu" -- exitWith below needs a real stage to unwind from
-    self:exitWith(Strings("Link error:\n%s", self.net.error or "?"))
+    self:exitWith(Strings("Link error:\n%s", detail))
   end
   return self
 end
@@ -163,21 +175,13 @@ end
 -- take the peer's hello out of the inbox without eating anything that
 -- shares the batch with it
 function LinkState:pollHello()
-  local msgs = self.net:poll()
-  local keep, got = {}, false
-  for _, msg in ipairs(msgs) do
-    if msg.type == "hello" and not self.peerHello then
-      self.peerHello = msg
-      self.peerName = msg.name
-      got = true
-    else
-      keep[#keep + 1] = msg
-    end
+  local message
+  if not self.peerHello then message = self.net:take("hello") end
+  if message then
+    self.peerHello = message
+    self.peerName = message.name
   end
-  for i = #keep, 1, -1 do
-    table.insert(self.net.inbox, 1, keep[i])
-  end
-  return got, #keep > 0
+  return message ~= nil, self.net:hasPending()
 end
 
 function LinkState:sendHello(mode)
@@ -220,13 +224,15 @@ function LinkState:update(dt)
   local input = self.game.input
   if self.net then
     self.net:update()
-    if self.net.error and not PRE_CONNECT_STAGES[self.stage] then
-      self:exitWith(Strings("Link error:\n%s", self.net.error:sub(1, 60)))
+    local status = self.net:getStatus()
+    if status == "failed" and not PRE_CONNECT_STAGES[self.stage] then
+      self:exitWith(Strings("Link error:\n%s",
+        (self.net.error or "?"):sub(1, 60)))
       return
     end
-    -- the peer vanished without a bye (only once the inbox is drained,
+    -- the peer vanished without a bye (only once the session FIFO drains,
     -- so a final message travelling with the disconnect still counts)
-    if self.net.closed and #self.net.inbox == 0
+    if status == "closed"
        and not PRE_CONNECT_STAGES[self.stage] and self.stage ~= "addrEntry"
        and self.stage ~= "codeEntry" and self.stage ~= "notice"
        and self.stage ~= "battleRunning" then
@@ -269,12 +275,15 @@ function LinkState:update(dt)
       self.stage = "menu"
       self.index = 1
     elseif input:wasPressed("a") then
-      self.net = Net.new()
       if self.index == 1 then
-        if self.net:host() then
+        local session, detail = openSession("host", function(transport)
+          return transport:host()
+        end)
+        if session then
+          self.net = session
           self.stage = "hosting"
         else
-          self:exitWith(Strings("Link error:\n%s", self.net.error or "?"))
+          self:exitWith(Strings("Link error:\n%s", detail))
         end
       else
         self.stage = "addrEntry"
@@ -289,11 +298,14 @@ function LinkState:update(dt)
       self.index = 2
     elseif input:wasPressed("a") then
       if self.index == 1 then
-        self.net = Net.new()
-        if self.net:hostOnline() then
+        local session, detail = openSession("host", function(transport)
+          return transport:hostOnline()
+        end)
+        if session then
+          self.net = session
           self.stage = "onlineHosting"
         else
-          self:exitWith(Strings("Link error:\n%s", self.net.error or "?"))
+          self:exitWith(Strings("Link error:\n%s", detail))
         end
       else
         self.stage = "codeEntry"
@@ -327,11 +339,14 @@ function LinkState:update(dt)
       CodeEntry.right(self.codeEntry)
     elseif input:wasPressed("a") then
       local code = CodeEntry.text(self.codeEntry)
-      self.net = Net.new()
-      if self.net:joinOnline(nil, code) then
+      local session, detail = openSession("guest", function(transport)
+        return transport:joinOnline(nil, code)
+      end)
+      if session then
+        self.net = session
         self.stage = "onlineJoining"
       else
-        self:exitWith(Strings("Link error:\n%s", self.net.error or "?"))
+        self:exitWith(Strings("Link error:\n%s", detail))
       end
     end
 
@@ -367,10 +382,15 @@ function LinkState:update(dt)
                                   + self.addr[base + 2] * 10
                                   + self.addr[base + 3])
       end
-      if self.net:join(table.concat(octets, ".")) then
+      local address = table.concat(octets, ".")
+      local session, detail = openSession("guest", function(transport)
+        return transport:join(address)
+      end)
+      if session then
+        self.net = session
         self.stage = "joining"
       else
-        self:exitWith(Strings("Link error:\n%s", self.net.error or "?"))
+        self:exitWith(Strings("Link error:\n%s", detail))
       end
     end
 
@@ -428,19 +448,11 @@ function LinkState:update(dt)
 
   elseif self.stage == "waitMode" then -- guest waits for host's pick
     if input:wasPressed("b") then self:exitWith(nil) return end
-    local msgs = self.net:poll()
-    for i, msg in ipairs(msgs) do
-      if msg.type == "hello" then
-        self.peerHello = msg
-        self.peerName = msg.name
-        -- the host's next messages (party, ...) can share this batch;
-        -- put them back so the new stage's poll sees them
-        for j = #msgs, i + 1, -1 do
-          table.insert(self.net.inbox, 1, msgs[j])
-        end
-        self:decideCompat(msg.mode, false)
-        break
-      end
+    local message = self.net:take("hello")
+    if message then
+      self.peerHello = message
+      self.peerName = message.name
+      self:decideCompat(message.mode, false)
     end
 
   elseif self.stage == "notice" then
@@ -456,40 +468,34 @@ function LinkState:update(dt)
 
   elseif self.stage == "battleWait" then
     if input:wasPressed("b") then self:exitWith(nil) return end
-    local msgs = self.net:poll()
-    for i, msg in ipairs(msgs) do
-      if msg.type == "party" then
-        -- the host owns this rule (same as mode); the guest only learns
-        -- it here, off the host's own party message
-        if not self.isHost then self.forceLevel = msg.forceLevel end
-        local LinkBattle = require("src.link.LinkBattle")
-        local opts = {
-          myParty = Protocol.packParty(self.game.save.party),
-          theirParty = msg.mons,
-          theirName = self.peerName or "FOE",
-          seed = self.isHost and self.linkSeed or msg.seed,
-          verdict = self.verdict,
-          strict = Handshake.strict(self.verdict),
-          forceLevel = self.forceLevel,
-        }
-        local battle, why
-        if self.isHost then
-          battle, why = LinkBattle.newHost(self.game, self.net, opts)
-        else
-          battle, why = LinkBattle.newGuest(self.game, self.net, opts)
-        end
-        if not battle then
-          self.net:send({ type = "bye" })
-          self:exitWith(why or Strings("Link battle\ncan't start."), "error")
-          return
-        end
-        self.game.stack:push(battle)
-        self.stage = "battleRunning"
-        for j = #msgs, i + 1, -1 do
-          table.insert(self.net.inbox, 1, msgs[j])
-        end
-        break
+    local message = self.net:take("party")
+    if message then
+      -- the host owns this rule (same as mode); the guest only learns
+      -- it here, off the host's own party message
+      if not self.isHost then self.forceLevel = message.forceLevel end
+      local LinkBattle = require("src.link.LinkBattle")
+      local opts = {
+        myParty = Protocol.packParty(self.game.save.party),
+        theirParty = message.mons,
+        theirName = self.peerName or "FOE",
+        seed = self.isHost and self.linkSeed or message.seed,
+        verdict = self.verdict,
+        strict = Handshake.strict(self.verdict),
+        forceLevel = self.forceLevel,
+      }
+      local battle, why
+      if self.isHost then
+        battle, why = LinkBattle.newHost(self.game, self.net, opts)
+      else
+        battle, why = LinkBattle.newGuest(self.game, self.net, opts)
       end
+      if not battle then
+        self.net:send({ type = "bye" })
+        self:exitWith(why or Strings("Link battle\ncan't start."), "error")
+        return
+      end
+      self.game.stack:push(battle)
+      self.stage = "battleRunning"
     end
 
   elseif self.stage == "battleRunning" then

src/link/Net.lua

@@ -207,7 +207,7 @@ function Net:send(msg)
   end
   if self.peerEnd then -- loopback: re-encode through json like the wire
     local decoded = Json.decode(Json.encode(msg))
-    if decoded and not self.peerEnd.closed then
+    if decoded ~= nil and not self.peerEnd.closed then
       table.insert(self.peerEnd.inbox, decoded)
     end
     return
@@ -256,13 +256,12 @@ end
 
 function Net:handleTCPLine(line)
   local msg = Json.decode(line)
-  if not msg then
+  if msg == nil then
     Logger.warn("link: bad relay message %q", line:sub(1, 60))
     return
   end
-  if not handleGenericRelayControl(self, msg) then
-    table.insert(self.inbox, msg)
-  end
+  if type(msg) == "table" and handleGenericRelayControl(self, msg) then return end
+  table.insert(self.inbox, msg)
 end
 
 -- pulls every complete "\n"-terminated line out of rxBuf (leaving a
@@ -352,7 +351,7 @@ function Net:update()
       end
     elseif event.type == "receive" then
       local msg = Json.decode(event.data)
-      if msg then
+      if msg ~= nil then
         table.insert(self.inbox, msg)
       else
         Logger.warn("link: bad message %q", tostring(event.data):sub(1, 60))

src/link/Session.lua

@@ -0,0 +1,198 @@
+local Session = {}
+Session.__index = Session
+
+local VALID_ROLES = { host = true, guest = true }
+local REQUIRED_METHODS = { "update", "poll", "send", "close" }
+
+function Session.new(transport, options)
+  assert(type(transport) == "table", "Session.new requires a transport")
+  assert(type(options) == "table", "Session.new requires options")
+  assert(VALID_ROLES[options.role], "Session role must be host or guest")
+  assert(type(options.kind) == "string" and options.kind ~= "",
+    "Session kind must be a non-empty string")
+  for _, method in ipairs(REQUIRED_METHODS) do
+    assert(type(transport[method]) == "function",
+      "Session transport requires " .. method)
+  end
+
+  local self = setmetatable({
+    _transport = transport,
+    _role = options.role,
+    _kind = options.kind,
+    _inbox = {},
+    _status = "connecting",
+    _terminal = nil,
+    _transportCloseCalled = false,
+    paired = false,
+    closed = false,
+    error = nil,
+    code = nil,
+    address = nil,
+    target = nil,
+  }, Session)
+  self:_syncMetadata()
+  self:_refreshStatus()
+  return self
+end
+
+function Session:_syncMetadata()
+  local transport = self._transport
+  self.paired = transport.paired == true
+  self.code = transport.code
+  self.address = transport.address
+  self.target = transport.target
+end
+
+function Session:_refreshStatus()
+  if not self._terminal then
+    self._status = self.paired and "paired" or "connecting"
+    self.closed = false
+    self.error = nil
+    return
+  end
+  if #self._inbox > 0 then
+    self._status = "draining"
+    self.closed = false
+    self.error = nil
+    return
+  end
+  self._status = self._terminal.status
+  self.closed = true
+  self.error = self._terminal.status == "failed"
+    and (self._terminal.detail or self._terminal.reason) or nil
+end
+
+function Session:_latchTerminal(status, reason, detail)
+  if self._terminal then return false end
+  self._terminal = { status = status, reason = reason, detail = detail }
+  self:_refreshStatus()
+  return true
+end
+
+function Session:_closeTransport()
+  if self._transportCloseCalled then return true end
+  self._transportCloseCalled = true
+  local ok, detail = pcall(self._transport.close, self._transport)
+  return ok, ok and nil or tostring(detail)
+end
+
+function Session:getRole() return self._role end
+function Session:getKind() return self._kind end
+function Session:getStatus() return self._status end
+function Session:getFailure()
+  if not self._terminal or self._terminal.status ~= "failed" then
+    return nil, nil
+  end
+  return self._terminal.reason, self._terminal.detail
+end
+function Session:hasPending() return #self._inbox > 0 end
+
+function Session:send(message)
+  if self._terminal then return nil end
+  return self._transport:send(message)
+end
+
+function Session:update()
+  if self._terminal then
+    self:_refreshStatus()
+    return
+  end
+
+  local failureReason, failureDetail
+  local updateOk, updateDetail = pcall(self._transport.update, self._transport)
+  self:_syncMetadata()
+  if not updateOk then
+    failureReason, failureDetail = "transport_error", tostring(updateDetail)
+  elseif self._transport.error then
+    failureReason = "transport_error"
+    failureDetail = tostring(self._transport.error)
+  end
+
+  local pollOk, messages = pcall(self._transport.poll, self._transport)
+  if not pollOk then
+    if not failureReason then
+      failureReason, failureDetail = "transport_error", tostring(messages)
+    end
+  elseif type(messages) ~= "table" then
+    if not failureReason then
+      failureReason, failureDetail = "transport_error",
+        "transport poll returned non-table"
+    end
+  else
+    for index = 1, #messages do
+      local message = messages[index]
+      if type(message) ~= "table" or type(message.type) ~= "string" then
+        if not failureReason then
+          failureReason = "protocol_error"
+          failureDetail = ("message %d must be a table with string type")
+            :format(index)
+        end
+        break
+      end
+      self._inbox[#self._inbox + 1] = message
+    end
+  end
+
+  self:_syncMetadata()
+  if failureReason then
+    self:_latchTerminal("failed", failureReason, failureDetail)
+    self:_closeTransport()
+  elseif self._transport.closed then
+    local closeOk, closeDetail = self:_closeTransport()
+    if closeOk then
+      self:_latchTerminal("closed")
+    else
+      self:_latchTerminal("failed", "transport_error", closeDetail)
+    end
+  end
+  self:_refreshStatus()
+end
+
+local function finishRead(self)
+  self:_refreshStatus()
+end
+
+function Session:take(messageType)
+  assert(type(messageType) == "string", "Session.take requires a message type")
+  for index, message in ipairs(self._inbox) do
+    if message.type == messageType then
+      local found = table.remove(self._inbox, index)
+      finishRead(self)
+      return found
+    end
+  end
+  return nil
+end
+
+function Session:pollOne()
+  if #self._inbox == 0 then return nil end
+  local message = table.remove(self._inbox, 1)
+  finishRead(self)
+  return message
+end
+
+function Session:poll()
+  local messages = self._inbox
+  self._inbox = {}
+  finishRead(self)
+  return messages
+end
+
+function Session:close()
+  if self._status == "closed" or self._status == "failed" then return end
+  if self._terminal then
+    self:_closeTransport()
+    self:_refreshStatus()
+    return
+  end
+  local ok, detail = self:_closeTransport()
+  if ok then
+    self:_latchTerminal("closed")
+  else
+    self:_latchTerminal("failed", "transport_error", detail)
+  end
+  self:_syncMetadata()
+  self:_refreshStatus()
+end
+
+return Session

src/link/Tournament.lua

@@ -14,6 +14,7 @@ local Font = require("src.render.Font")
 local Handshake = require("src.link.Handshake")
 local LinkBattle = require("src.link.LinkBattle")
 local Net = require("src.link.Net")
+local Session = require("src.link.Session")
 local Protocol = require("src.link.Protocol")
 local Runtime = require("src.mods.Runtime")
 local Sound = require("src.core.Sound")
@@ -132,12 +133,23 @@ end
 -- host / join
 -- -------------------------------------------------------------------
 
+local function openSession(role)
+  local transport = Net.new()
+  if transport:connectTCP(Net.defaultRelayAddress()) then
+    return Session.new(transport, { role = role, kind = "tournament" })
+  end
+  local detail = transport.error or "?"
+  transport:close()
+  return nil, detail
+end
+
 function Tournament:startHosting()
-  self.net = Net.new()
-  if not self.net:connectTCP(Net.defaultRelayAddress()) then
-    self:exitWith(Strings("Link error:\n%s", self.net.error or "?"))
+  local session, detail = openSession("host")
+  if not session then
+    self:exitWith(Strings("Link error:\n%s", detail))
     return
   end
+  self.net = session
   local size, minL, maxL = partyStats(self.game.save.party)
   self.isCreator = true
   self.participating = self.settings.participating
@@ -156,11 +168,12 @@ function Tournament:startHosting()
 end
 
 function Tournament:startJoining(code)
-  self.net = Net.new()
-  if not self.net:connectTCP(Net.defaultRelayAddress()) then
-    self:exitWith(Strings("Link error:\n%s", self.net.error or "?"))
+  local session, detail = openSession("guest")
+  if not session then
+    self:exitWith(Strings("Link error:\n%s", detail))
     return
   end
+  self.net = session
   local size, minL, maxL = partyStats(self.game.save.party)
   self.isCreator = false
   self.participating = true -- joining is always to compete; only hosting can opt out
@@ -256,22 +269,14 @@ function Tournament:sendHello(mode)
 end
 
 function Tournament:pollHello()
-  local msgs = self.net:poll()
-  local keep, got = {}, false
-  for _, msg in ipairs(msgs) do
-    if msg.type == "hello" and not self.peerHello then
-      self.peerHello = msg
-      got = true
-    else
-      keep[#keep + 1] = msg
-    end
-  end
-  for i = #keep, 1, -1 do
-    table.insert(self.net.inbox, 1, keep[i])
-  end
-  return got
+  local message
+  if not self.peerHello then message = self.net:take("hello") end
+  if message then self.peerHello = message end
+  return message ~= nil
 end
 
+-- The relay assigns a side for each match; this can differ from the immutable
+-- tournament creator/joiner role held by Session.
 function Tournament:enterMatch(msg)
   self.isHost = (msg.role == "host")
   self.opponentName = msg.opponent
@@ -356,12 +361,13 @@ function Tournament:update(dt)
 
   if self.net then
     self.net:update()
-    if self.net.error and self.stage ~= "menu" and self.stage ~= "hostSettings"
+    local status = self.net:getStatus()
+    if status == "failed" and self.stage ~= "menu" and self.stage ~= "hostSettings"
        and self.stage ~= "codeEntry" then
-      self:exitWith(Strings("Link error:\n%s", self.net.error:sub(1, 60)))
+      self:exitWith(Strings("Link error:\n%s", (self.net.error or "?"):sub(1, 60)))
       return
     end
-    if self.net.closed and self.stage ~= "done" then
+    if status == "closed" and self.stage ~= "done" then
       self:exitWith(Strings("The tournament\nconnection was\nlost."))
       return
     end
@@ -370,24 +376,23 @@ function Tournament:update(dt)
   if self.stage == "matchHello" then
     if input:wasPressed("b") then self:exitWith(nil) return end
     self:pollHello()
-    if self.peerHello then self:beginMatchBattle() end
-    for _, msg in ipairs(self.net:poll()) do self:handleMessage(msg) end
+    if self.peerHello then
+      self:beginMatchBattle()
+      return
+    end
+    for _, message in ipairs(self.net:poll()) do self:handleMessage(message) end
     return
   elseif self.stage == "matchWaitParty" then
     if input:wasPressed("b") then self:exitWith(nil) return end
-    local msgs = self.net:poll()
-    for i, msg in ipairs(msgs) do
-      if msg.type == "party" then
-        self.pendingBattleOpts.theirParty = msg.mons
+    while self.net:hasPending() do
+      local message = self.net:pollOne()
+      if message.type == "party" then
+        self.pendingBattleOpts.theirParty = message.mons
         if self.isHost then
           self.pendingBattleOpts.seed = self.pendingBattleOpts.seed or self.linkSeed
         else
-          self.pendingBattleOpts.seed = msg.seed
+          self.pendingBattleOpts.seed = message.seed
         end
-        -- Split rather than `cond and newHost() or newGuest()`: the and/or
-        -- idiom truncates a call to its first result, so the second return
-        -- (the specific reason) was always dropped and every failure showed
-        -- the generic fallback instead of "same mods on both games" etc.
         local battle, why
         if self.isHost then
           battle, why = LinkBattle.newHost(self.game, self.net, self.pendingBattleOpts)
@@ -398,27 +403,22 @@ function Tournament:update(dt)
           self:exitWith(why or Strings("Link battle\ncan't start."))
           return
         end
-        -- anything after `party` in this same batch belongs to the
-        -- battle now, not to Tournament -- put it back for its own poll()
-        for j = #msgs, i + 1, -1 do
-          table.insert(self.net.inbox, 1, msgs[j])
-        end
         self.activeBattle = battle
         self.game.stack:push(battle)
         self.stage = "matchRunning"
         return
       else
-        self:handleMessage(msg)
+        self:handleMessage(message)
       end
     end
     return
   elseif self.stage == "spectateWait" then
     if input:wasPressed("b") then self:exitWith(nil) return end
-    local msgs = self.net:poll()
-    for i, msg in ipairs(msgs) do
-      if msg.type == "spectate" and msg.msg.type == "party" then
-        local inner = msg.msg
-        if msg.side == "host" then
+    while self.net:hasPending() do
+      local message = self.net:pollOne()
+      if message.type == "spectate" and message.msg.type == "party" then
+        local inner = message.msg
+        if message.side == "host" then
           self.spectate.hostParty = inner.mons
           self.spectate.seed = inner.seed
         else
@@ -426,8 +426,10 @@ function Tournament:update(dt)
         end
         if self.spectate.hostParty and self.spectate.guestParty then
           local battle, why = LinkBattle.newSpectator(self.game, self.net, {
-            hostParty = self.spectate.hostParty, guestParty = self.spectate.guestParty,
-            hostName = self.spectate.hostName, guestName = self.spectate.guestName,
+            hostParty = self.spectate.hostParty,
+            guestParty = self.spectate.guestParty,
+            hostName = self.spectate.hostName,
+            guestName = self.spectate.guestName,
             seed = self.spectate.seed,
             forceLevel = levelForWire(self.settings.forceLevel),
           })
@@ -435,16 +437,13 @@ function Tournament:update(dt)
             self:exitWith(why or Strings("Can't watch this\nmatch."))
             return
           end
-          for j = #msgs, i + 1, -1 do
-            table.insert(self.net.inbox, 1, msgs[j])
-          end
           self.activeBattle = battle
           self.game.stack:push(battle)
           self.stage = "spectateRunning"
           return
         end
       else
-        self:handleMessage(msg)
+        self:handleMessage(message)
       end
     end
     return