summaryrefslogtreecommitdiff
path: root/main
diff options
context:
space:
mode:
Diffstat (limited to 'main')
-rw-r--r--main/Main.hs142
-rw-r--r--main/State.hs22
-rw-r--r--main/Terminal.hs5
-rw-r--r--main/Test.hs144
-rw-r--r--main/Test/Service.hs21
-rw-r--r--main/WebSocket.hs46
6 files changed, 288 insertions, 92 deletions
diff --git a/main/Main.hs b/main/Main.hs
index 3f78db1..403e5e9 100644
--- a/main/Main.hs
+++ b/main/Main.hs
@@ -11,6 +11,7 @@ import Control.Monad.Except
import Control.Monad.Reader
import Control.Monad.State
import Control.Monad.Trans.Maybe
+import Control.Monad.Writer
import Crypto.Random
@@ -61,6 +62,7 @@ import State
import Terminal
import Test
import Version
+import WebSocket
data Options = Options
{ optServer :: ServerOptions
@@ -68,6 +70,7 @@ data Options = Options
, optStorage :: StorageOption
, optChatroomAutoSubscribe :: Maybe Int
, optDmBotEcho :: Maybe Text
+ , optWebSocketServer :: Maybe Int
, optShowHelp :: Bool
, optShowVersion :: Bool
}
@@ -90,6 +93,7 @@ defaultOptions = Options
, optStorage = DefaultStorage
, optChatroomAutoSubscribe = Nothing
, optDmBotEcho = Nothing
+ , optWebSocketServer = Nothing
, optShowHelp = False
, optShowVersion = False
}
@@ -110,7 +114,7 @@ availableServices =
True "peer discovery"
]
-options :: [OptDescr (Options -> Options)]
+options :: [ OptDescr (Options -> Writer [ String ] Options) ]
options =
[ Option ['p'] ["port"]
(ReqArg (\p -> so $ \opts -> opts { serverPort = read p }) "<port>")
@@ -119,57 +123,81 @@ options =
(NoArg (so $ \opts -> opts { serverLocalDiscovery = False }))
"do not send announce packets for local discovery"
, Option [] [ "storage" ]
- (ReqArg (\path -> \opts -> opts { optStorage = FilesystemStorage path }) "<path>")
+ (ReqArg (\path -> \opts -> return opts { optStorage = FilesystemStorage path }) "<path>")
"use storage in <path>"
, Option [] [ "memory-storage" ]
- (NoArg (\opts -> opts { optStorage = MemoryStorage }))
+ (NoArg (\opts -> return opts { optStorage = MemoryStorage }))
"use memory storage"
, Option [] ["chatroom-auto-subscribe"]
- (ReqArg (\count -> \opts -> opts { optChatroomAutoSubscribe = Just (read count) }) "<count>")
+ (ReqArg (\count -> \opts -> return opts { optChatroomAutoSubscribe = Just (read count) }) "<count>")
"automatically subscribe for up to <count> chatrooms"
#ifdef ENABLE_ICE_SUPPORT
, Option [] [ "discovery-stun-port" ]
- (ReqArg (\value -> serviceAttr $ \attrs -> attrs { discoveryStunPort = Just (read value) }) "<port>")
+ (ReqArg (\value -> serviceAttr $ \attrs -> return attrs { discoveryStunPort = Just (read value) }) "<port>")
"offer specified <port> to discovery peers for STUN protocol"
, Option [] [ "discovery-stun-server" ]
- (ReqArg (\value -> serviceAttr $ \attrs -> attrs { discoveryStunServer = Just (read value) }) "<server>")
+ (ReqArg (\value -> serviceAttr $ \attrs -> return attrs { discoveryStunServer = Just (read value) }) "<server>")
"offer <server> (domain name or IP address) to discovery peers for STUN protocol"
, Option [] [ "discovery-turn-port" ]
- (ReqArg (\value -> serviceAttr $ \attrs -> attrs { discoveryTurnPort = Just (read value) }) "<port>")
+ (ReqArg (\value -> serviceAttr $ \attrs -> return attrs { discoveryTurnPort = Just (read value) }) "<port>")
"offer specified <port> to discovery peers for TURN protocol"
, Option [] [ "discovery-turn-server" ]
- (ReqArg (\value -> serviceAttr $ \attrs -> attrs { discoveryTurnServer = Just (read value) }) "<server>")
+ (ReqArg (\value -> serviceAttr $ \attrs -> return attrs { discoveryTurnServer = Just (read value) }) "<server>")
"offer <server> (domain name or IP address) to discovery peers for TURN protocol"
#endif
+ , Option [] [ "discovery-tunnel" ]
+ (OptArg (\value -> \opts -> do
+ fun <- provideTunnelFun value
+ serviceAttr (\attrs -> return attrs { discoveryProvideTunnel = fun }) opts) "<peer-type>")
+ "offer to provide tunnel for peers of given <peer-type>, possible values: all, none, websocket"
, Option [] ["dm-bot-echo"]
- (ReqArg (\prefix -> \opts -> opts { optDmBotEcho = Just (T.pack prefix) }) "<prefix>")
+ (ReqArg (\prefix -> \opts -> return opts { optDmBotEcho = Just (T.pack prefix) }) "<prefix>")
"automatically reply to direct messages with the same text prefixed with <prefix>"
+ , Option [] [ "websocket-server" ]
+ (ReqArg (\value -> \opts -> return opts { optWebSocketServer = Just (read value) }) "<port>")
+ "start WebSocket server on given port"
, Option ['h'] ["help"]
- (NoArg $ \opts -> opts { optShowHelp = True })
+ (NoArg $ \opts -> return opts { optShowHelp = True })
"show this help and exit"
, Option ['V'] ["version"]
- (NoArg $ \opts -> opts { optShowVersion = True })
+ (NoArg $ \opts -> return opts { optShowVersion = True })
"show version and exit"
]
where
- so f opts = opts { optServer = f $ optServer opts }
+ so f opts = return opts { optServer = f $ optServer opts }
- updateService :: Service s => (ServiceAttributes s -> ServiceAttributes s) -> SomeService -> SomeService
+ updateService :: (Service s, Monad m, Typeable m) => (ServiceAttributes s -> m (ServiceAttributes s)) -> SomeService -> m SomeService
updateService f some@(SomeService proxy attrs)
- | Just f' <- cast f = SomeService proxy (f' attrs)
- | otherwise = some
-
- serviceAttr :: Service s => (ServiceAttributes s -> ServiceAttributes s) -> Options -> Options
- serviceAttr f opts = opts { optServices = map (\sopt -> sopt { soptService = updateService f (soptService sopt) }) (optServices opts) }
-
-servicesOptions :: [OptDescr (Options -> Options)]
+ | Just f' <- cast f = SomeService proxy <$> f' attrs
+ | otherwise = return some
+
+ serviceAttr :: (Service s, Monad m, Typeable m) => (ServiceAttributes s -> m (ServiceAttributes s)) -> Options -> m Options
+ serviceAttr f opts = do
+ services' <- forM (optServices opts) $ \sopt -> do
+ service <- updateService f (soptService sopt)
+ return sopt { soptService = service }
+ return opts { optServices = services' }
+
+ provideTunnelFun :: Maybe String -> Writer [ String ] (Peer -> Bool)
+ provideTunnelFun Nothing = return $ const True
+ provideTunnelFun (Just "all") = return $ const True
+ provideTunnelFun (Just "none") = return $ const False
+ provideTunnelFun (Just "websocket") = return $ \peer ->
+ case peerAddress peer of
+ CustomPeerAddress addr | Just WebSocketAddress {} <- cast addr -> True
+ _ -> False
+ provideTunnelFun (Just name) = do
+ tell [ "Invalid value of --discovery-tunnel: ‘" <> name <> "’\n" ]
+ return $ const False
+
+servicesOptions :: [ OptDescr (Options -> Writer [ String ] Options) ]
servicesOptions = concatMap helper $ "all" : map soptName availableServices
where
helper name =
- [ Option [] ["enable-" <> name] (NoArg $ so $ change name $ \sopt -> sopt { soptEnabled = True }) ""
- , Option [] ["disable-" <> name] (NoArg $ so $ change name $ \sopt -> sopt { soptEnabled = False }) ""
+ [ Option [] [ "enable-" <> name ] (NoArg $ so $ change name $ \sopt -> sopt { soptEnabled = True }) ""
+ , Option [] [ "disable-" <> name ] (NoArg $ so $ change name $ \sopt -> sopt { soptEnabled = False }) ""
]
- so f opts = opts { optServices = f $ optServices opts }
+ so f opts = return opts { optServices = f $ optServices opts }
change :: String -> (ServiceOption -> ServiceOption) -> [ServiceOption] -> [ServiceOption]
change name f (s : ss)
| soptName s == name || name == "all"
@@ -187,13 +215,16 @@ getDefaultStorageDir = do
main :: IO ()
main = do
- (opts, args) <- (getOpt RequireOrder (options ++ servicesOptions) <$> getArgs) >>= \case
- (o, args, []) -> do
- return (foldl (flip id) defaultOptions o, args)
- (_, _, errs) -> do
+ let printErrors errs = do
progName <- getProgName
hPutStrLn stderr $ concat errs <> "Try `" <> progName <> " --help' for more information."
exitFailure
+ (opts, args) <- (getOpt RequireOrder (options ++ servicesOptions) <$> getArgs) >>= \case
+ (wo, args, []) ->
+ case runWriter (foldM (flip ($)) defaultOptions wo) of
+ ( o, [] ) -> return ( o, args )
+ ( _, errs ) -> printErrors errs
+ (_, _, errs) -> printErrors errs
st <- liftIO $ case optStorage opts of
DefaultStorage -> openStorage =<< getDefaultStorageDir
@@ -362,6 +393,10 @@ interactiveLoop st opts = withTerminal commandCompletion $ \term -> do
startServer (optServer opts) erebosHead extPrintLn $
map soptService $ filter soptEnabled $ optServices opts
+ case optWebSocketServer opts of
+ Just port -> startWebsocketServer server "::" port extPrintLn
+ Nothing -> return ()
+
void $ liftIO $ forkIO $ void $ forever $ do
peer <- getNextPeerChange server
peerIdentity peer >>= \case
@@ -493,7 +528,10 @@ getSelectedChatroom = gets csContext >>= \case
_ -> throwOtherError "no chatroom selected"
getSelectedConversation :: CommandM Conversation
-getSelectedConversation = gets csContext >>= \case
+getSelectedConversation = gets csContext >>= getConversationFromContext
+
+getConversationFromContext :: CommandContext -> CommandM Conversation
+getConversationFromContext = \case
SelectedPeer peer -> peerIdentity peer >>= \case
PeerIdentityFull pid -> directMessageConversation $ finalOwner pid
_ -> throwOtherError "incomplete peer identity"
@@ -507,6 +545,13 @@ getSelectedConversation = gets csContext >>= \case
SelectedConversation conv -> reloadConversation conv
_ -> throwOtherError "no contact, peer or conversation selected"
+getSelectedOrManualContext :: CommandM CommandContext
+getSelectedOrManualContext = do
+ asks ciLine >>= \case
+ "" -> gets csContext
+ str | all isDigit str -> getContextByIndex (read str)
+ _ -> throwOtherError "invalid index"
+
commands :: [(String, Command)]
commands =
[ ("history", cmdHistory)
@@ -628,19 +673,22 @@ cmdMembers = do
forM_ (chatroomMembers room) $ \x -> do
cmdPutStrLn $ maybe "<unnamed>" T.unpack $ idName x
+getContextByIndex :: Int -> CommandM CommandContext
+getContextByIndex n = do
+ join (asks ciContextOptions) >>= \ctxs -> if
+ | n > 0, (ctx : _) <- drop (n - 1) ctxs -> return ctx
+ | otherwise -> throwOtherError "invalid index"
cmdSelectContext :: Command
cmdSelectContext = do
n <- read <$> asks ciLine
- join (asks ciContextOptions) >>= \ctxs -> if
- | n > 0, (ctx : _) <- drop (n - 1) ctxs -> do
- modify $ \s -> s { csContext = ctx }
- case ctx of
- SelectedChatroom rstate -> do
- when (not (roomStateSubscribe rstate)) $ do
- chatroomSetSubscribe (head $ roomStateData rstate) True
- _ -> return ()
- | otherwise -> throwOtherError "invalid index"
+ ctx <- getContextByIndex n
+ modify $ \s -> s { csContext = ctx }
+ case ctx of
+ SelectedChatroom rstate -> do
+ when (not (roomStateSubscribe rstate)) $ do
+ chatroomSetSubscribe (head $ roomStateData rstate) True
+ _ -> return ()
cmdSend :: Command
cmdSend = void $ do
@@ -654,12 +702,12 @@ cmdSend = void $ do
cmdDelete :: Command
cmdDelete = void $ do
- deleteConversation =<< getSelectedConversation
+ deleteConversation =<< getConversationFromContext =<< getSelectedOrManualContext
modify $ \s -> s { csContext = NoContext }
cmdHistory :: Command
cmdHistory = void $ do
- conv <- getSelectedConversation
+ conv <- getConversationFromContext =<< getSelectedOrManualContext
case conversationHistory conv of
thread@(_:_) -> do
tzone <- liftIO $ getCurrentTimeZone
@@ -824,7 +872,7 @@ cmdConversations = do
cmdDetails :: Command
cmdDetails = do
- gets csContext >>= \case
+ getSelectedOrManualContext >>= \case
SelectedPeer peer -> do
cmdPutStrLn $ unlines
[ "Network peer:"
@@ -900,17 +948,11 @@ cmdDiscoveryInit = void $ do
cmdDiscovery :: Command
cmdDiscovery = void $ do
- Just peer <- gets csIcePeer
- st <- getStorage
+ server <- asks ciServer
sref <- asks ciLine
- eprint <- asks ciPrint
- liftIO $ readRef st (BC.pack sref) >>= \case
- Nothing -> error "ref does not exist"
- Just ref -> do
- res <- runExceptT $ sendToPeer peer $ DiscoverySearch ref
- case res of
- Right _ -> return ()
- Left err -> eprint err
+ case readRefDigest (BC.pack sref) of
+ Nothing -> throwOtherError "failed to parse ref"
+ Just dgst -> discoverySearch server dgst
#ifdef ENABLE_ICE_SUPPORT
diff --git a/main/State.hs b/main/State.hs
index 76441df..150178e 100644
--- a/main/State.hs
+++ b/main/State.hs
@@ -44,22 +44,22 @@ loadLocalStateHead term st = loadHeads st >>= \case
, ssValue = [ storedRef $ idExtData $ fromMaybe identity owner ]
}
storeHead st $ LocalState
- { lsIdentity = idExtData identity
+ { lsPrev = Nothing
+ , lsIdentity = idExtData identity
, lsShared = [ shared ]
, lsOther = []
}
updateSharedIdentity :: (MonadHead LocalState m, MonadError e m, FromErebosError e) => Terminal -> m ()
-updateSharedIdentity term = updateLocalHead_ $ updateSharedState_ $ \case
+updateSharedIdentity term = updateLocalState_ $ updateSharedState_ $ \case
Just identity -> do
Just . toComposedIdentity <$> interactiveIdentityUpdate term identity
Nothing -> throwOtherError "no existing shared identity"
interactiveIdentityUpdate :: (Foldable f, MonadStorage m, MonadIO m, MonadError e m, FromErebosError e) => Terminal -> Identity f -> m UnifiedIdentity
-interactiveIdentityUpdate term identity = do
- let public = idKeyIdentity identity
-
+interactiveIdentityUpdate term fidentity = do
+ identity <- mergeIdentity fidentity
name <- liftIO $ do
setPrompt term $ T.unpack $ T.concat $ concat
[ [ T.pack "Name" ]
@@ -70,11 +70,11 @@ interactiveIdentityUpdate term identity = do
]
getInputLine term $ KeepPrompt . maybe T.empty T.pack
- if | T.null name -> mergeIdentity identity
+ if | T.null name -> return identity
| otherwise -> do
- secret <- loadKey public
- maybe (throwOtherError "created invalid identity") return . validateIdentity =<<
- mstore =<< sign secret =<< mstore (emptyIdentityData public)
- { iddPrev = toList $ idDataF identity
- , iddName = Just name
+ secret <- loadKey $ idKeyIdentity identity
+ maybe (throwOtherError "created invalid identity") return . validateExtendedIdentity =<<
+ mstore =<< sign secret =<< mstore . ExtendedIdentityData =<< return (emptyIdentityExtension $ idData identity)
+ { idePrev = toList $ idExtDataF identity
+ , ideName = Just name
}
diff --git a/main/Terminal.hs b/main/Terminal.hs
index 5dc3612..150bd8c 100644
--- a/main/Terminal.hs
+++ b/main/Terminal.hs
@@ -31,8 +31,9 @@ import Data.List
import Data.Text (Text)
import Data.Text qualified as T
-import System.IO
import System.Console.ANSI
+import System.IO
+import System.IO.Error
data Terminal = Terminal
@@ -107,7 +108,7 @@ termPutStr Terminal {..} str = do
getInput :: IO Input
getInput = do
- getChar >>= \case
+ handleJust (guard . isEOFError) (\() -> return InputEnd) $ getChar >>= \case
'\ESC' -> do
esc <- readEsc
case parseEsc esc of
diff --git a/main/Test.hs b/main/Test.hs
index 08ad880..b07bd87 100644
--- a/main/Test.hs
+++ b/main/Test.hs
@@ -26,7 +26,7 @@ import Data.Text qualified as T
import Data.Text.Encoding
import Data.Text.IO qualified as T
import Data.Typeable
-import Data.UUID qualified as U
+import Data.UUID.Types qualified as U
import Network.Socket
@@ -44,6 +44,7 @@ import Erebos.Object
import Erebos.Pairing
import Erebos.PubKey
import Erebos.Service
+import Erebos.Service.Stream
import Erebos.Set
import Erebos.State
import Erebos.Storable
@@ -66,10 +67,17 @@ data TestState = TestState
data RunningServer = RunningServer
{ rsServer :: Server
- , rsPeers :: MVar (Int, [(Int, Peer)])
+ , rsPeers :: MVar ( Int, [ TestPeer ] )
, rsPeerThread :: ThreadId
}
+data TestPeer = TestPeer
+ { tpIndex :: Int
+ , tpPeer :: Peer
+ , tpStreamReaders :: MVar [ (Int, StreamReader ) ]
+ , tpStreamWriters :: MVar [ (Int, StreamWriter ) ]
+ }
+
initTestState :: TestState
initTestState = TestState
{ tsHead = Nothing
@@ -137,17 +145,20 @@ cmdOut line = do
getPeer :: Text -> CommandM Peer
-getPeer spidx = do
+getPeer spidx = tpPeer <$> getTestPeer spidx
+
+getTestPeer :: Text -> CommandM TestPeer
+getTestPeer spidx = do
Just RunningServer {..} <- gets tsServer
- Just peer <- lookup (read $ T.unpack spidx) . snd <$> liftIO (readMVar rsPeers)
+ Just peer <- find (((read $ T.unpack spidx) ==) . tpIndex) . snd <$> liftIO (readMVar rsPeers)
return peer
-getPeerIndex :: MVar (Int, [(Int, Peer)]) -> ServiceHandler (PairingService a) Int
+getPeerIndex :: MVar ( Int, [ TestPeer ] ) -> ServiceHandler s Int
getPeerIndex pmvar = do
peer <- asks svcPeer
- maybe 0 fst . find ((==peer) . snd) . snd <$> liftIO (readMVar pmvar)
+ maybe 0 tpIndex . find ((peer ==) . tpPeer) . snd <$> liftIO (readMVar pmvar)
-pairingAttributes :: PairingResult a => proxy (PairingService a) -> Output -> MVar (Int, [(Int, Peer)]) -> String -> PairingAttributes a
+pairingAttributes :: PairingResult a => proxy (PairingService a) -> Output -> MVar ( Int, [ TestPeer ] ) -> String -> PairingAttributes a
pairingAttributes _ out peers prefix = PairingAttributes
{ pairingHookRequest = return ()
@@ -216,6 +227,11 @@ directMessageAttributes out = DirectMessageAttributes
{ dmOwnerMismatch = afterCommit $ outLine out "dm-owner-mismatch"
}
+discoveryAttributes :: DiscoveryAttributes
+discoveryAttributes = (defaultServiceAttributes Proxy)
+ { discoveryProvideTunnel = const True
+ }
+
dmReceivedWatcher :: Output -> Stored DirectMessage -> IO ()
dmReceivedWatcher out smsg = do
let msg = fromStored smsg
@@ -267,6 +283,9 @@ commands = map (T.pack *** id)
, ("peer-drop", cmdPeerDrop)
, ("peer-list", cmdPeerList)
, ("test-message-send", cmdTestMessageSend)
+ , ("test-stream-open", cmdTestStreamOpen)
+ , ("test-stream-close", cmdTestStreamClose)
+ , ("test-stream-send", cmdTestStreamSend)
, ("local-state-get", cmdLocalStateGet)
, ("local-state-replace", cmdLocalStateReplace)
, ("local-state-wait", cmdLocalStateWait)
@@ -286,6 +305,7 @@ commands = map (T.pack *** id)
, ("contact-set-name", cmdContactSetName)
, ("dm-send-peer", cmdDmSendPeer)
, ("dm-send-contact", cmdDmSendContact)
+ , ("dm-send-identity", cmdDmSendIdentity)
, ("dm-list-peer", cmdDmListPeer)
, ("dm-list-contact", cmdDmListContact)
, ("chatroom-create", cmdChatroomCreate)
@@ -301,6 +321,7 @@ commands = map (T.pack *** id)
, ("chatroom-leave", cmdChatroomLeave)
, ("chatroom-message-send", cmdChatroomMessageSend)
, ("discovery-connect", cmdDiscoveryConnect)
+ , ("discovery-tunnel", cmdDiscoveryTunnel)
]
cmdStore :: Command
@@ -454,7 +475,8 @@ cmdCreateIdentity = do
_ -> return []
storeHead st $ LocalState
- { lsIdentity = idExtData identity
+ { lsPrev = Nothing
+ , lsIdentity = idExtData identity
, lsShared = shared
, lsOther = []
}
@@ -493,14 +515,27 @@ cmdStartServer = do
"attach" -> return $ someServiceAttr $ pairingAttributes (Proxy @AttachService) out rsPeers "attach"
"chatroom" -> return $ someService @ChatroomService Proxy
"contact" -> return $ someServiceAttr $ pairingAttributes (Proxy @ContactService) out rsPeers "contact"
- "discovery" -> return $ someService @DiscoveryService Proxy
+ "discovery" -> return $ someServiceAttr $ discoveryAttributes
"dm" -> return $ someServiceAttr $ directMessageAttributes out
"sync" -> return $ someService @SyncService Proxy
"test" -> return $ someServiceAttr $ (defaultServiceAttributes Proxy)
{ testMessageReceived = \obj otype len sref -> do
liftIO $ do
void $ store (headStorage h) obj
- outLine out $ unwords ["test-message-received", otype, len, sref]
+ outLine out $ unwords [ "test-message-received", otype, len, sref ]
+ , testStreamsReceived = \streams -> do
+ pidx <- getPeerIndex rsPeers
+ liftIO $ do
+ nums <- mapM getStreamReaderNumber streams
+ outLine out $ unwords $ "test-stream-open-from" : show pidx : map show nums
+ forM_ (zip nums streams) $ \( num, stream ) -> void $ forkIO $ do
+ let go = readStreamPacket stream >>= \case
+ StreamData seqNum bytes -> do
+ outLine out $ unwords [ "test-stream-received", show pidx, show num, show seqNum, BC.unpack bytes ]
+ go
+ StreamClosed seqNum -> do
+ outLine out $ unwords [ "test-stream-closed-from", show pidx, show num, show seqNum ]
+ go
}
sname -> throwOtherError $ "unknown service `" <> T.unpack sname <> "'"
@@ -509,15 +544,23 @@ cmdStartServer = do
rsPeerThread <- liftIO $ forkIO $ void $ forever $ do
peer <- getNextPeerChange rsServer
- let printPeer (idx, p) = do
- params <- peerIdentity p >>= return . \case
+ let printPeer TestPeer {..} = do
+ params <- peerIdentity tpPeer >>= return . \case
PeerIdentityFull pid -> ("id":) $ map (maybe "<unnamed>" T.unpack . idName) (unfoldOwners pid)
- _ -> [ "addr", show (peerAddress p) ]
- outLine out $ unwords $ [ "peer", show idx ] ++ params
+ _ -> [ "addr", show (peerAddress tpPeer) ]
+ outLine out $ unwords $ [ "peer", show tpIndex ] ++ params
+
+ update ( tpIndex, [] ) = do
+ tpPeer <- return peer
+ tpStreamReaders <- newMVar []
+ tpStreamWriters <- newMVar []
+ let tp = TestPeer {..}
+ printPeer tp
+ return ( tpIndex + 1, [ tp ] )
- update (nid, []) = printPeer (nid, peer) >> return (nid + 1, [(nid, peer)])
- update cur@(nid, p:ps) | snd p == peer = printPeer p >> return cur
- | otherwise = fmap (p:) <$> update (nid, ps)
+ update cur@( nid, p : ps )
+ | tpPeer p == peer = printPeer p >> return cur
+ | otherwise = fmap (p :) <$> update ( nid, ps )
modifyMVar_ rsPeers update
@@ -554,10 +597,10 @@ cmdPeerList = do
peers <- liftIO $ getCurrentPeerList rsServer
tpeers <- liftIO $ readMVar rsPeers
forM_ peers $ \peer -> do
- Just (n, _) <- return $ find ((peer==).snd) . snd $ tpeers
+ Just tp <- return $ find ((peer ==) . tpPeer) . snd $ tpeers
mbpid <- peerIdentity peer
cmdOut $ unwords $ concat
- [ [ "peer-list-item", show n ]
+ [ [ "peer-list-item", show (tpIndex tp) ]
, [ "addr", show (peerAddress peer) ]
, case mbpid of PeerIdentityFull pid -> ("id":) $ map (maybe "<unnamed>" T.unpack . idName) (unfoldOwners pid)
_ -> []
@@ -574,6 +617,40 @@ cmdTestMessageSend = do
sendManyToPeer peer $ map (TestMessage . wrappedLoad) refs
cmdOut "test-message-send done"
+cmdTestStreamOpen :: Command
+cmdTestStreamOpen = do
+ spidx : rest <- asks tiParams
+ tp <- getTestPeer spidx
+ count <- case rest of
+ [] -> return 1
+ tcount : _ -> return $ read $ T.unpack tcount
+
+ out <- asks tiOutput
+ runPeerService (tpPeer tp) $ do
+ streams <- openTestStreams count
+ afterCommit $ do
+ nums <- mapM getStreamWriterNumber streams
+ modifyMVar_ (tpStreamWriters tp) $ return . (++ zip nums streams)
+ outLine out $ unwords $ "test-stream-open-done"
+ : T.unpack spidx
+ : map show nums
+
+cmdTestStreamClose :: Command
+cmdTestStreamClose = do
+ [ spidx, sid ] <- asks tiParams
+ tp <- getTestPeer spidx
+ Just stream <- lookup (read $ T.unpack sid) <$> liftIO (readMVar (tpStreamWriters tp))
+ liftIO $ closeStream stream
+ cmdOut $ unwords [ "test-stream-close-done", T.unpack spidx, T.unpack sid ]
+
+cmdTestStreamSend :: Command
+cmdTestStreamSend = do
+ [ spidx, sid, content ] <- asks tiParams
+ tp <- getTestPeer spidx
+ Just stream <- lookup (read $ T.unpack sid) <$> liftIO (readMVar (tpStreamWriters tp))
+ liftIO $ writeStream stream $ encodeUtf8 content
+ cmdOut $ unwords [ "test-stream-send-done", T.unpack spidx, T.unpack sid ]
+
cmdLocalStateGet :: Command
cmdLocalStateGet = do
h <- getHead
@@ -646,7 +723,7 @@ cmdWatchSharedIdentity = do
cmdUpdateLocalIdentity :: Command
cmdUpdateLocalIdentity = do
[name] <- asks tiParams
- updateLocalHead_ $ \ls -> do
+ updateLocalState_ $ \ls -> do
Just identity <- return $ validateExtendedIdentity $ lsIdentity $ fromStored ls
let public = idKeyIdentity identity
@@ -661,7 +738,7 @@ cmdUpdateLocalIdentity = do
cmdUpdateSharedIdentity :: Command
cmdUpdateSharedIdentity = do
[name] <- asks tiParams
- updateLocalHead_ $ updateSharedState_ $ \case
+ updateLocalState_ $ updateSharedState_ $ \case
Nothing -> throwOtherError "no existing shared identity"
Just identity -> do
let public = idKeyIdentity identity
@@ -731,7 +808,7 @@ cmdContactSetName :: Command
cmdContactSetName = do
[cid, name] <- asks tiParams
contact <- getContact cid
- updateLocalHead_ $ updateSharedState_ $ contactSetName contact name
+ updateLocalState_ $ updateSharedState_ $ contactSetName contact name
cmdOut "contact-set-name-done"
cmdDmSendPeer :: Command
@@ -746,6 +823,14 @@ cmdDmSendContact = do
Just to <- contactIdentity <$> getContact cid
void $ sendDirectMessage to msg
+cmdDmSendIdentity :: Command
+cmdDmSendIdentity = do
+ st <- asks tiStorage
+ [ tid, msg ] <- asks tiParams
+ Just ref <- liftIO $ readRef st $ encodeUtf8 tid
+ Just to <- return $ validateExtendedIdentity $ wrappedLoad ref
+ void $ sendDirectMessage to msg
+
dmList :: Foldable f => Identity f -> Command
dmList peer = do
threads <- toThreadList . lookupSharedValue . lsShared . headObject <$> getHead
@@ -887,11 +972,14 @@ cmdChatroomMessageSend = do
cmdDiscoveryConnect :: Command
cmdDiscoveryConnect = do
- st <- asks tiStorage
[ tref ] <- asks tiParams
- Just ref <- liftIO $ readRef st $ encodeUtf8 tref
-
+ Just dgst <- return $ readRefDigest $ encodeUtf8 tref
Just RunningServer {..} <- gets tsServer
- peers <- liftIO $ getCurrentPeerList rsServer
- forM_ peers $ \peer -> do
- sendToPeer peer $ DiscoverySearch ref
+ discoverySearch rsServer dgst
+
+cmdDiscoveryTunnel :: Command
+cmdDiscoveryTunnel = do
+ [ tvia, ttarget ] <- asks tiParams
+ via <- getPeer tvia
+ Just target <- return $ readRefDigest $ encodeUtf8 ttarget
+ liftIO $ discoverySetupTunnel via target
diff --git a/main/Test/Service.hs b/main/Test/Service.hs
index 8c58dee..c0be07d 100644
--- a/main/Test/Service.hs
+++ b/main/Test/Service.hs
@@ -1,8 +1,11 @@
module Test.Service (
TestMessage(..),
TestMessageAttributes(..),
+
+ openTestStreams,
) where
+import Control.Monad
import Control.Monad.Reader
import Data.ByteString.Lazy.Char8 qualified as BL
@@ -10,12 +13,14 @@ import Data.ByteString.Lazy.Char8 qualified as BL
import Erebos.Network
import Erebos.Object
import Erebos.Service
+import Erebos.Service.Stream
import Erebos.Storable
data TestMessage = TestMessage (Stored Object)
data TestMessageAttributes = TestMessageAttributes
{ testMessageReceived :: Object -> String -> String -> String -> ServiceHandler TestMessage ()
+ , testStreamsReceived :: [ StreamReader ] -> ServiceHandler TestMessage ()
}
instance Storable TestMessage where
@@ -26,7 +31,10 @@ instance Service TestMessage where
serviceID _ = mkServiceID "cb46b92c-9203-4694-8370-8742d8ac9dc8"
type ServiceAttributes TestMessage = TestMessageAttributes
- defaultServiceAttributes _ = TestMessageAttributes (\_ _ _ _ -> return ())
+ defaultServiceAttributes _ = TestMessageAttributes
+ { testMessageReceived = \_ _ _ _ -> return ()
+ , testStreamsReceived = \_ -> return ()
+ }
serviceHandler smsg = do
let TestMessage sobj = fromStored smsg
@@ -36,3 +44,14 @@ instance Service TestMessage where
cb <- asks $ testMessageReceived . svcAttributes
cb obj otype len (show $ refDigest $ storedRef sobj)
_ -> return ()
+
+ streams <- receivedStreams
+ when (not $ null streams) $ do
+ cb <- asks $ testStreamsReceived . svcAttributes
+ cb streams
+
+
+openTestStreams :: Int -> ServiceHandler TestMessage [ StreamWriter ]
+openTestStreams count = do
+ replyPacket . TestMessage =<< mstore (Rec [])
+ replicateM count openStream
diff --git a/main/WebSocket.hs b/main/WebSocket.hs
new file mode 100644
index 0000000..5c49aa1
--- /dev/null
+++ b/main/WebSocket.hs
@@ -0,0 +1,46 @@
+module WebSocket (
+ WebSocketAddress(..),
+ startWebsocketServer,
+) where
+
+import Control.Concurrent
+import Control.Exception
+import Control.Monad
+
+import Data.ByteString.Lazy qualified as BL
+import Data.Unique
+
+import Erebos.Network
+
+import Network.WebSockets qualified as WS
+
+
+data WebSocketAddress = WebSocketAddress Unique WS.Connection
+
+instance Eq WebSocketAddress where
+ WebSocketAddress u _ == WebSocketAddress u' _ = u == u'
+
+instance Ord WebSocketAddress where
+ compare (WebSocketAddress u _) (WebSocketAddress u' _) = compare u u'
+
+instance Show WebSocketAddress where
+ show (WebSocketAddress _ _) = "websocket"
+
+instance PeerAddressType WebSocketAddress where
+ sendBytesToAddress (WebSocketAddress _ conn) msg = do
+ WS.sendDataMessage conn $ WS.Binary $ BL.fromStrict msg
+
+startWebsocketServer :: Server -> String -> Int -> (String -> IO ()) -> IO ()
+startWebsocketServer server addr port logd = do
+ void $ forkIO $ do
+ WS.runServer addr port $ \pending -> do
+ conn <- WS.acceptRequest pending
+ u <- newUnique
+ let paddr = WebSocketAddress u conn
+ void $ serverPeerCustom server paddr
+ handle (\(e :: SomeException) -> logd $ "WebSocket thread exception: " ++ show e) $ do
+ WS.withPingThread conn 30 (return ()) $ do
+ forever $ do
+ WS.receiveDataMessage conn >>= \case
+ WS.Binary msg -> receivedFromCustomAddress server paddr $ BL.toStrict msg
+ WS.Text {} -> logd $ "unexpected websocket text message"
generated by cgit v1.2.3 (git 2.25.1) at 2025-07-10 14:57:29 +0000