module Network.LibP2P.Protocol.GossipSub.MessageCache
( newMessageCache
, cachePut
, cacheGet
, cacheGetGossipIds
, cacheShift
) where
import Data.Foldable (toList)
import qualified Data.Map.Strict as Map
import qualified Data.Sequence as Seq
import Network.LibP2P.Protocol.GossipSub.Types
( Topic, MessageId, PubSubMessage (..)
, CacheEntry (..), MessageCache (..)
)
newMessageCache :: Int -> Int -> MessageCache
newMessageCache :: Int -> Int -> MessageCache
newMessageCache Int
len Int
gossip = MessageCache
{ mcWindows :: Seq [CacheEntry]
mcWindows = Int -> [CacheEntry] -> Seq [CacheEntry]
forall a. Int -> a -> Seq a
Seq.replicate Int
len []
, mcIndex :: Map MessageId CacheEntry
mcIndex = Map MessageId CacheEntry
forall k a. Map k a
Map.empty
, mcLen :: Int
mcLen = Int
len
, mcGossip :: Int
mcGossip = Int
gossip
}
cachePut :: MessageId -> PubSubMessage -> MessageCache -> MessageCache
cachePut :: MessageId -> PubSubMessage -> MessageCache -> MessageCache
cachePut MessageId
mid PubSubMessage
msg MessageCache
mc =
let entry :: CacheEntry
entry = MessageId -> PubSubMessage -> Topic -> CacheEntry
CacheEntry MessageId
mid PubSubMessage
msg (PubSubMessage -> Topic
msgTopic PubSubMessage
msg)
windows :: Seq [CacheEntry]
windows = case Seq [CacheEntry] -> ViewL [CacheEntry]
forall a. Seq a -> ViewL a
Seq.viewl (MessageCache -> Seq [CacheEntry]
mcWindows MessageCache
mc) of
ViewL [CacheEntry]
Seq.EmptyL -> [CacheEntry] -> Seq [CacheEntry]
forall a. a -> Seq a
Seq.singleton [CacheEntry
entry]
[CacheEntry]
newest Seq.:< Seq [CacheEntry]
rest -> (CacheEntry
entry CacheEntry -> [CacheEntry] -> [CacheEntry]
forall a. a -> [a] -> [a]
: [CacheEntry]
newest) [CacheEntry] -> Seq [CacheEntry] -> Seq [CacheEntry]
forall a. a -> Seq a -> Seq a
Seq.<| Seq [CacheEntry]
rest
in MessageCache
mc
{ mcWindows = windows
, mcIndex = Map.insert mid entry (mcIndex mc)
}
cacheGet :: MessageId -> MessageCache -> Maybe PubSubMessage
cacheGet :: MessageId -> MessageCache -> Maybe PubSubMessage
cacheGet MessageId
mid MessageCache
mc = CacheEntry -> PubSubMessage
ceMessage (CacheEntry -> PubSubMessage)
-> Maybe CacheEntry -> Maybe PubSubMessage
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> MessageId -> Map MessageId CacheEntry -> Maybe CacheEntry
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup MessageId
mid (MessageCache -> Map MessageId CacheEntry
mcIndex MessageCache
mc)
cacheGetGossipIds :: Topic -> MessageCache -> [MessageId]
cacheGetGossipIds :: Topic -> MessageCache -> [MessageId]
cacheGetGossipIds Topic
topic MessageCache
mc =
let gossipWindows :: Seq [CacheEntry]
gossipWindows = Int -> Seq [CacheEntry] -> Seq [CacheEntry]
forall a. Int -> Seq a -> Seq a
Seq.take (MessageCache -> Int
mcGossip MessageCache
mc) (MessageCache -> Seq [CacheEntry]
mcWindows MessageCache
mc)
entries :: [CacheEntry]
entries = ([CacheEntry] -> [CacheEntry]) -> [[CacheEntry]] -> [CacheEntry]
forall (t :: * -> *) a b. Foldable t => (a -> [b]) -> t a -> [b]
concatMap [CacheEntry] -> [CacheEntry]
forall a. a -> a
id (Seq [CacheEntry] -> [[CacheEntry]]
forall a. Seq a -> [a]
forall (t :: * -> *) a. Foldable t => t a -> [a]
toList Seq [CacheEntry]
gossipWindows)
in [ CacheEntry -> MessageId
ceMessageId CacheEntry
e | CacheEntry
e <- [CacheEntry]
entries, CacheEntry -> Topic
ceTopic CacheEntry
e Topic -> Topic -> Bool
forall a. Eq a => a -> a -> Bool
== Topic
topic ]
cacheShift :: MessageCache -> MessageCache
cacheShift :: MessageCache -> MessageCache
cacheShift MessageCache
mc =
let windows :: Seq [CacheEntry]
windows = MessageCache -> Seq [CacheEntry]
mcWindows MessageCache
mc
(Seq [CacheEntry]
kept, [CacheEntry]
dropped) = case Seq [CacheEntry] -> ViewR [CacheEntry]
forall a. Seq a -> ViewR a
Seq.viewr Seq [CacheEntry]
windows of
ViewR [CacheEntry]
Seq.EmptyR -> (Seq [CacheEntry]
forall a. Seq a
Seq.empty, [])
Seq [CacheEntry]
rest Seq.:> [CacheEntry]
oldest -> (Seq [CacheEntry]
rest, [CacheEntry]
oldest)
newWindows :: Seq [CacheEntry]
newWindows = (Seq [CacheEntry]
forall a. Seq a
Seq.empty Seq [CacheEntry] -> [CacheEntry] -> Seq [CacheEntry]
forall a. Seq a -> a -> Seq a
Seq.|> []) Seq [CacheEntry] -> Seq [CacheEntry] -> Seq [CacheEntry]
forall a. Seq a -> Seq a -> Seq a
Seq.>< Seq [CacheEntry]
kept
droppedIds :: [MessageId]
droppedIds = (CacheEntry -> MessageId) -> [CacheEntry] -> [MessageId]
forall a b. (a -> b) -> [a] -> [b]
map CacheEntry -> MessageId
ceMessageId [CacheEntry]
dropped
newIndex :: Map MessageId CacheEntry
newIndex = (Map MessageId CacheEntry -> MessageId -> Map MessageId CacheEntry)
-> Map MessageId CacheEntry
-> [MessageId]
-> Map MessageId CacheEntry
forall b a. (b -> a -> b) -> b -> [a] -> b
forall (t :: * -> *) b a.
Foldable t =>
(b -> a -> b) -> b -> t a -> b
foldl' (\Map MessageId CacheEntry
idx MessageId
mid -> MessageId -> Map MessageId CacheEntry -> Map MessageId CacheEntry
forall k a. Ord k => k -> Map k a -> Map k a
Map.delete MessageId
mid Map MessageId CacheEntry
idx) (MessageCache -> Map MessageId CacheEntry
mcIndex MessageCache
mc) [MessageId]
droppedIds
in MessageCache
mc
{ mcWindows = newWindows
, mcIndex = newIndex
}