add lease time to subscription entries

src/Hash2Pub/PostService.hs

@@ -60,7 +60,8 @@ type RelayTags = RingMap NodeID (TagSubscribersSTM, TChan PostID, Hashtag)
 type TagSubscribersSTM = TVar TagSubscribers
 -- | each subscriber is identified by its contact data "hostname" "port"
 -- and holds a TChan duplicated from the broadcast TChan of the tag
-type TagSubscribers = (HMap.HashMap (String, Int) (TChan PostID))
+-- + an expiration timestamp
+type TagSubscribers = (HMap.HashMap (String, Int) (TChan PostID, POSIXTime))
 
 
 instance DHT d => Service PostService d where
@@ -175,10 +176,12 @@ subscriptionDelivery serv subList = do
       processTag :: TVar RelayTags -> Txt.Text -> IO ()
       processTag subscriberSTM tagData = do
           let
-            tag:subText:posts:_ = Txt.splitOn "," tagData
+            tag:subText:lease:posts:_ = Txt.splitOn "," tagData
+            -- ignore checking of lease time
+            leaseTime = fromIntegral (read . Txt.unpack $ lease :: Integer)
             sub = read . Txt.unpack $ subText :: (String, Int)
             postList = Txt.words posts
-          enqueueSubscriptions subscriberSTM (normaliseTag tag) sub postList
+          enqueueSubscription subscriberSTM (normaliseTag tag) sub postList leaseTime
 
 
 postFetch :: PostService d -> Txt.Text -> Handler Txt.Text
@@ -232,39 +235,43 @@ tagUnsubscribe serv hashtag origin = pure $ "Here be a dragon unsubscription fro
 
 -- | Write all pending posts of a subscriber-tag-combination to its queue.
 -- Sets up all necessary data structures if they are still missing.
-enqueueSubscriptions :: TVar RelayTags    -- tag-subscriber map
+enqueueSubscription :: TVar RelayTags    -- tag-subscriber map
                      -> Hashtag         -- hashtag of pending posts
                      -> (String, Int)   -- subscriber's connection information
                      -> [PostID]        -- pending posts
+                     -> POSIXTime       -- lease expiry time
                      -> IO ()
-enqueueSubscriptions tagMapSTM tag subscriber posts = do
+enqueueSubscription tagMapSTM tag subscriber posts leaseTime = do
     -- get the tag output queue and, if necessary, create it
-    subChan <- atomically setupSubscriberChannel
+    subChan <- atomically $ setupSubscriberChannel tagMapSTM tag subscriber leaseTime
     forM_ posts (atomically . writeTChan subChan)
-  where
+
-      setupSubscriberChannel :: STM (TChan PostID)
+
-      setupSubscriberChannel = do
+-- | STM operation to return the outgoing post queue of a tag to a specified subscriber.
-          tagMap <- readTVar tagMapSTM
+-- If the queue doesn't exist yet, all necessary data structures are set up accordingly.
-          case lookupRelayTags tag tagMap of
+setupSubscriberChannel :: TVar RelayTags -> Hashtag -> (String, Int) -> POSIXTime -> STM (TChan PostID)
-            Nothing -> do
+setupSubscriberChannel tagMapSTM tag subscriber leaseTime = do
-              -- if no collision/ tag doesn't exist yet, just initialize a
+    tagMap <- readTVar tagMapSTM
-              -- new subscriber map
+    case lookupRelayTags tag tagMap of
-              broadcastChan <- newBroadcastTChan
+      Nothing -> do
-              tagOutChan <- dupTChan broadcastChan
+        -- if no collision/ tag doesn't exist yet, just initialize a
-              newSubMapSTM <- newTVar $ HMap.singleton subscriber tagOutChan
+        -- new subscriber map
-              writeTVar tagMapSTM $ addRMapEntry (genKeyID . Txt.unpack $ tag) (newSubMapSTM, broadcastChan, tag) tagMap
+        broadcastChan <- newBroadcastTChan
-              pure tagOutChan
+        tagOutChan <- dupTChan broadcastChan
-            Just (foundSubMapSTM, broadcastChan, _) -> do
+        newSubMapSTM <- newTVar $ HMap.singleton subscriber (tagOutChan, leaseTime)
-              -- otherwise use the existing subscriber map
+        writeTVar tagMapSTM $ addRMapEntry (genKeyID . Txt.unpack $ tag) (newSubMapSTM, broadcastChan, tag) tagMap
-              foundSubMap <- readTVar foundSubMapSTM
+        pure tagOutChan
-              case HMap.lookup subscriber foundSubMap of
+      Just (foundSubMapSTM, broadcastChan, _) -> do
-                Nothing -> do
+        -- otherwise use the existing subscriber map
-                  -- for new subscribers, create new output channel
+        foundSubMap <- readTVar foundSubMapSTM
-                  tagOutChan <- dupTChan broadcastChan
+        case HMap.lookup subscriber foundSubMap of
-                  writeTVar foundSubMapSTM $ HMap.insert subscriber tagOutChan foundSubMap
+          Nothing -> do
-                  pure tagOutChan
+            -- for new subscribers, create new output channel
-                -- existing subscriber's channels are just returned
+            tagOutChan <- dupTChan broadcastChan
-                Just tagOutChan -> pure tagOutChan
+            writeTVar foundSubMapSTM $ HMap.insert subscriber (tagOutChan, leaseTime) foundSubMap
+            pure tagOutChan
+          -- existing subscriber's channels are just returned
+          Just (tagOutChan, _) -> pure tagOutChan
 
 
 -- | returns the broadcast channel of a hashtag if there are any subscribers to it