schmittlauch/Hash2Pub
1
0
Fork 1
Code Issues 47 Pull requests Releases Wiki Activity

signal and handle non-responsibility to subscriptions

Browse source
This commit is contained in:
Trolli Schmittlauch 2020-08-13 23:44:24 +02:00
parent e646045ab2
commit 402378a78b
1 changed files with 23 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

34
src/Hash2Pub/PostService.hs
View file

@ -272,8 +272,13 @@ tagDelivery serv hashtag posts = do
pure () pure ()
pure $ "Received a postID for tag " <> hashtag pure $ "Received a postID for tag " <> hashtag
tagSubscribe :: PostService d -> Txt.Text -> Maybe Txt.Text -> Handler Integer tagSubscribe :: DHT d => PostService d -> Txt.Text -> Maybe Txt.Text -> Handler Integer
tagSubscribe serv hashtag origin = do tagSubscribe serv hashtag origin = do
responsible <- liftIO $ isResponsibleFor (baseDHT serv) (genKeyID . Txt.unpack $ hashtag)
if not responsible
-- GONE if not responsible
then throwError err410 { errBody = "not responsible for this tag" }
else pure ()
originURL <- maybe originURL <- maybe
(throwError $ err400 { errBody = "Missing Origin header" }) (throwError $ err400 { errBody = "Missing Origin header" })
pure pure
@ -359,16 +364,23 @@ clientDeliverSubscriptions serv fromTag toTag (toHost, toPort) = do
clientSubscribeTo :: DHT d => PostService d -> Hashtag -> IO (Either String Integer) clientSubscribeTo :: DHT d => PostService d -> Hashtag -> IO (Either String Integer)
clientSubscribeTo serv tag = do clientSubscribeTo serv tag = do
lookupRes <- lookupKey (baseDHT serv) (Txt.unpack tag) lookupRes <- lookupKey (baseDHT serv) (Txt.unpack tag)
let origin = "http://" <> Txt.pack (confServiceHost $ serviceConf serv) <> ":" <> Txt.pack (show (getListeningPortFromService serv :: Integer)) doSubscribe lookupRes True
maybe where
(pure . Left $ "No node found") doSubscribe lookupResponse allowRetry = maybe
(\(foundHost, foundPort) -> do (pure . Left $ "No node found")
resp <- runClientM (tagSubscribeClient tag (Just origin)) (mkClientEnv (httpMan serv) (BaseUrl Http foundHost (fromIntegral foundPort) "")) (\(foundHost, foundPort) -> do
case resp of let origin = "http://" <> Txt.pack (confServiceHost $ serviceConf serv) <> ":" <> Txt.pack (show (getListeningPortFromService serv :: Integer))
Left err -> pure . Left . show $ err resp <- runClientM (tagSubscribeClient tag (Just origin)) (mkClientEnv (httpMan serv) (BaseUrl Http foundHost (fromIntegral foundPort) ""))
Right lease -> pure . Right $ lease case resp of
) Left (FailureResponse _ fresp)
lookupRes |(HTTPT.statusCode . responseStatusCode $ fresp) == 410 && allowRetry -> do -- responsibility gone, force new lookup
newRes <- forceLookupKey (baseDHT serv) (Txt.unpack tag)
doSubscribe newRes False
Left err -> pure . Left . show $ err
Right lease -> pure . Right $ lease
)
lookupResponse
-- currently this is unused code -- currently this is unused code
getClients :: String -> Int -> HTTP.Manager -> Client IO PostServiceAPI getClients :: String -> Int -> HTTP.Manager -> Client IO PostServiceAPI