diff --git a/Network/Haskoin/Wallet/Client/Commands.hs b/Network/Haskoin/Wallet/Client/Commands.hs --- a/Network/Haskoin/Wallet/Client/Commands.hs +++ b/Network/Haskoin/Wallet/Client/Commands.hs @@ -96,13 +96,13 @@ cmdStart = do cfg <- R.ask liftIO $ runSPVServer cfg - liftIO $ putStrLn "Process started" -- hw stop [config] cmdStop :: Handler () -cmdStop = R.ask >>= \cfg -> liftIO $ do - stopSPVServer cfg - putStrLn "Process stopped" +cmdStop = do + resE <- sendZmq StopServerR + handleResponse (resE :: Either String (WalletResponse (Maybe ()))) (const $ return ()) + liftIO $ putStrLn "Process stopped" getSigningKeys :: String -> Handler (Maybe XPrvKey) diff --git a/Network/Haskoin/Wallet/Server.hs b/Network/Haskoin/Wallet/Server.hs --- a/Network/Haskoin/Wallet/Server.hs +++ b/Network/Haskoin/Wallet/Server.hs @@ -20,6 +20,7 @@ void, when) import Control.Monad.Base (MonadBase) import Control.Monad.Catch (MonadThrow) +import Control.Monad.Fix (fix) import Control.Monad.Logger (MonadLoggerIO, filterLogger, logDebug, logError, logInfo, @@ -120,8 +121,17 @@ case configMode cfg of -- In this mode, we do not launch an SPV node. We only accept -- client requests through the ZMQ API. - SPVOffline -> - runWalletApp ctx $ HandlerSession cfg pool Nothing notif + SPVOffline -> do + let session = HandlerSession cfg pool Nothing notif + as <- mapM async + -- Run the ZMQ API-command server + [ runWalletCmd ctx session + -- Run the ZMQ notification thread + , runWalletNotif ctx session + ] + mapM_ link as + (_,r) <- waitAnyCancel as + return r -- In this mode, we launch the client ZMQ API and we sync the -- wallet database with an SPV node. SPVOnline -> do @@ -140,11 +150,14 @@ , runNodeT (handleGetData $ (`runDBPool` pool) . getTx) node -- Re-broadcast pending transactions , runNodeT (broadcastPendingTxs pool) node - -- Run the ZMQ API server - , runWalletApp ctx session + -- Run the ZMQ API-command server + , runWalletCmd ctx session + -- Run the ZMQ notification thread + , runWalletNotif ctx session ] mapM_ link as (_,r) <- waitAnyCancel as + $(logDebug) "Exiting main thread" return r where spv pool = do @@ -171,6 +184,7 @@ -- Start the merkle sync merkleSync pool 500 notif + $(logDebug) "Exiting Merkle-sync thread" -- Run a thread that will re-broadcast pending transactions broadcastPendingTxs pool = forever $ do (hash, _) <- runSqlNodeT $ walletBestBlock @@ -184,16 +198,19 @@ atomicallyNodeT $ do synced <- areBlocksSynced hash when synced $ lift retry - processTx pool notif = awaitForever $ \tx -> lift $ do - (_, newAddrs) <- runDBPool (importNetTx tx (Just notif)) pool - unless (null newAddrs) $ do - $(logInfo) $ pack $ unwords - [ "Generated", show $ length newAddrs - , "new addresses while importing the tx." - , "Updating the bloom filter" - ] - (bloom, elems, _) <- runDBPool getBloomFilter pool - atomicallyNodeT $ sendBloomFilter bloom elems + $(logDebug) "Exiting tx-broadcast thread" + processTx pool notif = do + awaitForever $ \tx -> lift $ do + (_, newAddrs) <- runDBPool (importNetTx tx (Just notif)) pool + unless (null newAddrs) $ do + $(logInfo) $ pack $ unwords + [ "Generated", show $ length newAddrs + , "new addresses while importing the tx." + , "Updating the bloom filter" + ] + (bloom, elems, _) <- runDBPool getBloomFilter pool + atomicallyNodeT $ sendBloomFilter bloom elems + $(logDebug) "Exiting tx-import thread" initDatabase :: (MonadBaseControl IO m, MonadLoggerIO m) => Config -> m ConnectionPool @@ -332,8 +349,9 @@ maybeDetach :: Config -> IO () -> IO () maybeDetach cfg action = - if configDetach cfg then runDetached pidFile logFile action else action + if configDetach cfg then runDetached pidFile logFile action >> logStarted else action where + logStarted = putStrLn "Process started" pidFile = Just $ configPidFile cfg logFile = ToFile $ configLogFile cfg @@ -345,17 +363,17 @@ -- Run the main ZeroMQ loop -- TODO: Support concurrent requests using DEALER socket when we can do -- concurrent MySQL requests. -runWalletApp :: ( MonadLoggerIO m +runWalletNotif :: ( MonadLoggerIO m , MonadBaseControl IO m , MonadBase IO m , MonadThrow m , MonadResource m ) - => Context -> HandlerSession -> m () -runWalletApp ctx session = do - na <- async $ liftBaseOpDiscard (withSocket ctx Pub) $ \sock -> do + => Context -> HandlerSession -> m () +runWalletNotif ctx session = + liftBaseOpDiscard (withSocket ctx Pub) $ \sock -> do liftIO $ setLinger (restrict (0 :: Int)) sock - setupCrypto ctx sock + setupCrypto ctx sock session liftIO $ bind sock $ configBindNotif $ handlerConfig session forever $ do xM <- liftIO $ atomically $ readTBMChan $ handlerNotifChan session @@ -366,32 +384,33 @@ NotifTx JsonTx{..} -> ("{" <> cs jsonTxAccount <> "}", cs $ encode x) in liftIO $ sendMulti sock $ typ :| [pay] - link na + +runWalletCmd :: ( MonadLoggerIO m + , MonadBaseControl IO m + , MonadBase IO m + , MonadThrow m + , MonadResource m + ) + => Context -> HandlerSession -> m () +runWalletCmd ctx session = do liftBaseOpDiscard (withSocket ctx Rep) $ \sock -> do liftIO $ setLinger (restrict (0 :: Int)) sock - setupCrypto ctx sock + setupCrypto ctx sock session liftIO $ bind sock $ configBind $ handlerConfig session - forever $ do + fix $ \loop -> do bs <- liftIO $ receive sock - res <- case decode $ BL.fromStrict bs of + let msg = decode $ BL.fromStrict bs + res <- case msg of + Just StopServerR -> do + $(logInfo) "Received StopServer request" + return (ResponseValid Nothing) Just r -> catchErrors $ runHandler (dispatchRequest r) session Nothing -> return $ ResponseError "Could not decode request" liftIO $ send sock [] $ BL.toStrict $ encode res + unless (msg == Just StopServerR) loop + $(logInfo) "Exiting ZMQ command thread..." where - setupCrypto :: (MonadLoggerIO m, MonadBaseControl IO m) - => Context -> Socket a -> m () - setupCrypto ctx' sock = do - when (isJust serverKeyM) $ liftIO $ do - let k = fromJust $ configServerKey $ handlerConfig session - setCurveServer True sock - setCurveSecretKey TextFormat k sock - when (isJust clientKeyPubM) $ do - k <- z85Decode (fromJust clientKeyPubM) - void $ async $ runZapAuth ctx' k - cfg = handlerConfig session - serverKeyM = configServerKey cfg - clientKeyPubM = configClientKeyPub cfg catchErrors m = catches m [ E.Handler $ \(WalletException err) -> do $(logError) $ pack err @@ -404,6 +423,21 @@ return $ ResponseError $ pack $ show exc ] +setupCrypto :: (MonadLoggerIO m, MonadBaseControl IO m) + => Context -> Socket a -> HandlerSession -> m () +setupCrypto ctx' sock session = do + when (isJust serverKeyM) $ liftIO $ do + let k = fromJust $ configServerKey $ handlerConfig session + setCurveServer True sock + setCurveSecretKey TextFormat k sock + when (isJust clientKeyPubM) $ do + k <- z85Decode (fromJust clientKeyPubM) + void $ async $ runZapAuth ctx' k + where + cfg = handlerConfig session + serverKeyM = configServerKey cfg + clientKeyPubM = configClientKeyPub cfg + runZapAuth :: ( MonadLoggerIO m , MonadBaseControl IO m , MonadBase IO m @@ -447,29 +481,30 @@ ) => WalletRequest -> Handler m (WalletResponse Value) dispatchRequest req = fmap ResponseValid $ case req of - GetAccountsR p -> getAccountsR p - PostAccountsR na -> postAccountsR na - PostAccountRenameR n n' -> postAccountRenameR n n' - GetAccountR n -> getAccountR n - PostAccountKeysR n ks -> postAccountKeysR n ks - PostAccountGapR n g -> postAccountGapR n g - GetAddressesR n t m o p -> getAddressesR n t m o p - GetAddressesUnusedR n t p -> getAddressesUnusedR n t p - GetAddressR n i t m o -> getAddressR n i t m o - GetIndexR n k t -> getIndexR n k t - PutAddressR n i t l -> putAddressR n i t l - PostAddressesR n i t -> postAddressesR n i t - GetTxsR n p -> getTxsR n p - GetAddrTxsR n i t p -> getAddrTxsR n i t p - PostTxsR n k a -> postTxsR n k a - GetTxR n h -> getTxR n h - GetOfflineTxR n h -> getOfflineTxR n h - PostOfflineTxR n k t c -> postOfflineTxR n k t c - GetBalanceR n mc o -> getBalanceR n mc o - PostNodeR na -> postNodeR na - DeleteTxIdR t -> deleteTxIdR t - GetSyncR a n b -> getSyncR a (Right n) b - GetSyncHeightR a n b -> getSyncR a (Left n) b - GetPendingR a p -> getPendingR a p - GetDeadR a p -> getDeadR a p - GetBlockInfoR l -> getBlockInfoR l + GetAccountsR p -> getAccountsR p + PostAccountsR na -> postAccountsR na + PostAccountRenameR n n' -> postAccountRenameR n n' + GetAccountR n -> getAccountR n + PostAccountKeysR n ks -> postAccountKeysR n ks + PostAccountGapR n g -> postAccountGapR n g + GetAddressesR n t m o p -> getAddressesR n t m o p + GetAddressesUnusedR n t p -> getAddressesUnusedR n t p + GetAddressR n i t m o -> getAddressR n i t m o + GetIndexR n k t -> getIndexR n k t + PutAddressR n i t l -> putAddressR n i t l + PostAddressesR n i t -> postAddressesR n i t + GetTxsR n p -> getTxsR n p + GetAddrTxsR n i t p -> getAddrTxsR n i t p + PostTxsR n k a -> postTxsR n k a + GetTxR n h -> getTxR n h + GetOfflineTxR n h -> getOfflineTxR n h + PostOfflineTxR n k t c -> postOfflineTxR n k t c + GetBalanceR n mc o -> getBalanceR n mc o + PostNodeR na -> postNodeR na + DeleteTxIdR t -> deleteTxIdR t + GetSyncR a n b -> getSyncR a (Right n) b + GetSyncHeightR a n b -> getSyncR a (Left n) b + GetPendingR a p -> getPendingR a p + GetDeadR a p -> getDeadR a p + GetBlockInfoR l -> getBlockInfoR l + StopServerR -> error "Should be handled elsewhere" diff --git a/Network/Haskoin/Wallet/Types.hs b/Network/Haskoin/Wallet/Types.hs --- a/Network/Haskoin/Wallet/Types.hs +++ b/Network/Haskoin/Wallet/Types.hs @@ -353,6 +353,8 @@ | GetPendingR !AccountName !ListRequest | GetDeadR !AccountName !ListRequest | GetBlockInfoR ![BlockHash] + | StopServerR + deriving (Show, Eq) -- TODO: Set omitEmptyContents on aeson-0.9 $(deriveJSON diff --git a/haskoin-wallet.cabal b/haskoin-wallet.cabal --- a/haskoin-wallet.cabal +++ b/haskoin-wallet.cabal @@ -1,5 +1,5 @@ name: haskoin-wallet -version: 0.4.1 +version: 0.4.2 synopsis: Implementation of a Bitcoin SPV Wallet with BIP32 and multisig support. description: @@ -111,6 +111,7 @@ ghc-options: -Wall -O3 -threaded + -eventlog -rtsopts -with-rtsopts=-N4