Service.run -> Daemon.run

This commit is contained in:
Lex Berezhny 2020-05-20 18:05:13 -04:00
parent d11f4f9bed
commit 7ffb169376
4 changed files with 37 additions and 32 deletions

View file

@ -1,11 +1,10 @@
import os
import asyncio
import logging
import signal
from typing import List, Optional, Tuple, NamedTuple
from aiohttp.web import GracefulExit
from lbry.conf import Config
from lbry.db import Database
from lbry.db.constants import TXO_TYPES
from lbry.schema.result import Censor
@ -23,8 +22,10 @@ class BlockEvent(NamedTuple):
class Sync:
def __init__(self, service: 'Service'):
self.service = service
def __init__(self, ledger: Ledger, db: Database):
self.ledger = ledger
self.conf = ledger.conf
self.db = db
self._on_block_controller = EventController()
self.on_block = self._on_block_controller.stream
@ -74,30 +75,6 @@ class Service:
self._on_connected_controller = EventController()
self.on_connected = self._on_connected_controller.stream
def run(self):
loop = asyncio.get_event_loop()
def exit():
raise GracefulExit()
try:
loop.add_signal_handler(signal.SIGINT, exit)
loop.add_signal_handler(signal.SIGTERM, exit)
except NotImplementedError:
pass # Not implemented on Windows
try:
loop.run_until_complete(self.start())
loop.run_forever()
except (GracefulExit, KeyboardInterrupt, asyncio.CancelledError):
pass
finally:
loop.run_until_complete(self.stop())
logging.shutdown()
if hasattr(loop, 'shutdown_asyncgens'):
loop.run_until_complete(loop.shutdown_asyncgens())
async def start(self):
await self.db.open()
await self.wallets.ensure_path_exists()

View file

@ -1,14 +1,16 @@
import json
import asyncio
import logging
import signal
from weakref import WeakSet
from aiohttp.web import GracefulExit
from aiohttp.web import Application, AppRunner, WebSocketResponse, TCPSite, Response
from aiohttp.http_websocket import WSMsgType, WSCloseCode
from lbry.extras.daemon.json_response_encoder import JSONResponseEncoder
from lbry.service.base import Service
from lbry.service.api import API
from lbry.console import Console
def jsonrpc_dumps_pretty(obj, **kwargs):
@ -62,9 +64,10 @@ class WebSocketManager(WebSocketResponse):
class Daemon:
def __init__(self, service: Service):
def __init__(self, service: Service, console: Console):
self.service = service
self.conf = service.conf
self.console = console
self.api = API(service)
self.app = Application()
self.app['websockets'] = WeakSet()
@ -80,7 +83,32 @@ class Daemon:
self.app.on_shutdown.append(self.on_shutdown)
self.runner = AppRunner(self.app)
def run(self):
loop = asyncio.get_event_loop()
def exit():
raise GracefulExit()
try:
loop.add_signal_handler(signal.SIGINT, exit)
loop.add_signal_handler(signal.SIGTERM, exit)
except NotImplementedError:
pass # Not implemented on Windows
try:
loop.run_until_complete(self.start())
loop.run_forever()
except (GracefulExit, KeyboardInterrupt, asyncio.CancelledError):
pass
finally:
loop.run_until_complete(self.stop())
logging.shutdown()
if hasattr(loop, 'shutdown_asyncgens'):
loop.run_until_complete(loop.shutdown_asyncgens())
async def start(self):
self.console.starting()
await self.runner.setup()
site = TCPSite(self.runner, 'localhost', self.conf.api_port)
await site.start()

View file

@ -19,7 +19,7 @@ class FullNode(Service):
def __init__(self, ledger: Ledger, db_url: str, chain: Lbrycrd = None):
super().__init__(ledger, db_url)
self.chain = chain or Lbrycrd(ledger)
self.sync = BlockchainSync(self, self.chain)
self.sync = BlockchainSync(self.chain, self.db, self.conf.processes)
async def start(self):
await self.chain.open()

View file

@ -1,12 +1,12 @@
import logging
from lbry.conf import Config
from lbry.service.api import Client
from lbry.blockchain.ledger import Ledger
from lbry.db import Database
from lbry.wallet.sync import SPVSync
from .base import Service
from .api import Client
log = logging.getLogger(__name__)