mirror of
https://github.com/yzernik/squeaknode.git
synced 2026-08-17 13:07:32 +02:00
Upload squeaks to server (#13)
* Include address in squeak database. * Add tables to db for uploader * Include address in squeak database. * Add tables to db for uploader * Add uploader class. * Fix foreign key relation in client db. * Change uploader to check all squeaks based on address and block number. * Start and stop uploader thread in client.
This commit is contained in:
Jonathan Zernik
GitHub
parent
890fb68c7c
commit
7ae6ea1f08
6 changed files with 221 additions and 24 deletions
|
|
@ -1,12 +1,15 @@
|
|||
import logging
|
||||
import threading
|
||||
|
||||
from squeak.core.signing import CSigningKey
|
||||
from squeak.core.signing import CSqueakAddress
|
||||
|
||||
from squeaknode.client.squeak_store import SqueakStore
|
||||
from squeaknode.common.blockchain_client import BlockchainClient
|
||||
from squeaknode.common.lightning_client import LightningClient
|
||||
from squeaknode.common.squeak_maker import SqueakMaker
|
||||
from squeaknode.client.db import SQLiteDBFactory
|
||||
from squeaknode.client.uploader import Uploader
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
|
@ -26,10 +29,22 @@ class SqueakNodeClient(object):
|
|||
self.blockchain_client = blockchain_client
|
||||
self.lightning_client = lightning_client
|
||||
self.signing_key = signing_key
|
||||
self.address = CSqueakAddress.from_verifying_key(signing_key.get_verifying_key())
|
||||
self.squeak_store = SqueakStore(db_factory)
|
||||
self.hub_store = None
|
||||
|
||||
# Event is set when the client stops
|
||||
self.stopped = threading.Event()
|
||||
self.uploader = Uploader(self.hub_store, self.squeak_store, self.address, self.stopped)
|
||||
|
||||
def start(self):
|
||||
self.uploader.start()
|
||||
|
||||
def stop(self):
|
||||
self.stopped.set()
|
||||
|
||||
def get_address(self):
|
||||
pass
|
||||
return self.address
|
||||
|
||||
def make_squeak(self, content):
|
||||
if self.signing_key is None:
|
||||
|
|
@ -38,11 +53,8 @@ class SqueakNodeClient(object):
|
|||
squeak_maker = SqueakMaker(self.signing_key, self.blockchain_client)
|
||||
squeak = squeak_maker.make_squeak(content)
|
||||
logger.info('Made squeak: {}'.format(squeak))
|
||||
self.add_squeak(squeak)
|
||||
return squeak
|
||||
|
||||
def add_squeak(self, squeak):
|
||||
self.squeak_store.save_squeak(squeak)
|
||||
return squeak
|
||||
|
||||
def get_squeak(self, squeak_hash):
|
||||
return self.squeak_store.get_squeak(squeak_hash)
|
||||
|
|
|
|||
55
squeaknode/client/hub_store.py
Normal file
55
squeaknode/client/hub_store.py
Normal file
|
|
@ -0,0 +1,55 @@
|
|||
import logging
|
||||
|
||||
from squeak.core import CSqueak
|
||||
from squeak.core import CheckSqueak
|
||||
from squeak.core import HASH_LENGTH
|
||||
from squeak.core import MakeSqueakFromStr
|
||||
from squeak.core import CSqueakEncContent
|
||||
from squeak.core.signing import CSigningKey
|
||||
from squeak.core.signing import CSigningKey
|
||||
from squeak.core.script import CScript
|
||||
|
||||
from squeaknode.client.db import get_db
|
||||
from squeaknode.client.db import close_db
|
||||
from squeaknode.client.db import initialize_db
|
||||
from squeaknode.client.db import SQLiteDBFactory
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class HubStore(object):
|
||||
"""Network node that handles client commands.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
db_factory: SQLiteDBFactory,
|
||||
) -> None:
|
||||
self.db_factory = db_factory
|
||||
|
||||
def save_hub(self, hub):
|
||||
host, port = hub
|
||||
with self.db_factory.make_conn() as conn:
|
||||
conn.execute(
|
||||
"INSERT INTO hub (host, port) VALUES (?, ?)",
|
||||
(host, port),
|
||||
)
|
||||
|
||||
def get_hubs(self):
|
||||
with self.db_factory.make_conn() as conn:
|
||||
hub_rows = (
|
||||
conn
|
||||
.execute(
|
||||
"SELECT host, port"
|
||||
" FROM hub h",
|
||||
)
|
||||
.fetchall()
|
||||
)
|
||||
hubs = []
|
||||
for hub_row in hub_rows:
|
||||
hubs.append(
|
||||
hub_row['host'],
|
||||
hub_row['port'],
|
||||
)
|
||||
return hubs
|
||||
|
|
@ -141,12 +141,13 @@ def run_client(config):
|
|||
signal.signal(signal.SIGTERM, sigterm_handler)
|
||||
|
||||
print("Starting client...")
|
||||
node.start()
|
||||
try:
|
||||
while True:
|
||||
time.sleep(1)
|
||||
finally:
|
||||
print("Shutting down...")
|
||||
close_db(db)
|
||||
node.stop()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
|
|
|||
|
|
@ -18,6 +18,24 @@ CREATE TABLE squeak (
|
|||
nNonce INTEGER NOT NULL,
|
||||
encContent BLOB NOT NULL,
|
||||
scriptSig BLOB NOT NULL,
|
||||
address TEXT NOT NULL,
|
||||
vchDataKey TEXT,
|
||||
content BLOB
|
||||
);
|
||||
|
||||
|
||||
CREATE TABLE hub (
|
||||
host TEXT NOT NULL,
|
||||
port TEXT NOT NULL,
|
||||
created TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
PRIMARY KEY(host, port)
|
||||
);
|
||||
|
||||
|
||||
CREATE TABLE upload (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
created TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
squeakHash TEXT NOT NULL,
|
||||
complete INTEGER NOT NULL,
|
||||
FOREIGN KEY (squeakHash) REFERENCES [squeak] (hash) ON UPDATE NO ACTION ON DELETE CASCADE
|
||||
);
|
||||
|
|
|
|||
|
|
@ -28,11 +28,33 @@ class SqueakStore(object):
|
|||
) -> None:
|
||||
self.db_factory = db_factory
|
||||
|
||||
def _row_to_squeak(self, squeak_row):
|
||||
squeak = CSqueak(
|
||||
nVersion=squeak_row['nVersion'],
|
||||
hashEncContent=squeak_row['hashEncContent'],
|
||||
hashReplySqk=squeak_row['hashReplySqk'],
|
||||
hashBlock=squeak_row['hashBlock'],
|
||||
nBlockHeight=squeak_row['nBlockHeight'],
|
||||
scriptPubKey=CScript(squeak_row['scriptPubKey']),
|
||||
hashDataKey=squeak_row['hashDataKey'],
|
||||
vchIv=squeak_row['vchIv'],
|
||||
nTime=squeak_row['nTime'],
|
||||
nNonce=squeak_row['nNonce'],
|
||||
encContent=CSqueakEncContent(squeak_row['encContent']),
|
||||
scriptSig=CScript(squeak_row['scriptSig']),
|
||||
vchDataKey=squeak_row['vchDataKey'],
|
||||
)
|
||||
try:
|
||||
CheckSqueak(squeak)
|
||||
return squeak
|
||||
except:
|
||||
return None
|
||||
|
||||
def save_squeak(self, squeak):
|
||||
CheckSqueak(squeak)
|
||||
with self.db_factory.make_conn() as conn:
|
||||
conn.execute(
|
||||
"INSERT INTO squeak (hash, nVersion, hashEncContent, hashReplySqk, hashBlock, nBlockHeight, scriptPubKey, hashDataKey, vchIv, nTime, nNonce, encContent, scriptSig, vchDataKey, content) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||||
"INSERT INTO squeak (hash, nVersion, hashEncContent, hashReplySqk, hashBlock, nBlockHeight, scriptPubKey, hashDataKey, vchIv, nTime, nNonce, encContent, scriptSig, address, vchDataKey, content) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||||
(
|
||||
squeak.GetHash(),
|
||||
squeak.nVersion,
|
||||
|
|
@ -47,6 +69,7 @@ class SqueakStore(object):
|
|||
squeak.nNonce,
|
||||
bytes(squeak.encContent.vchEncContent),
|
||||
bytes(squeak.scriptSig),
|
||||
str(squeak.GetAddress()),
|
||||
squeak.vchDataKey,
|
||||
"",
|
||||
),
|
||||
|
|
@ -66,26 +89,58 @@ class SqueakStore(object):
|
|||
)
|
||||
if squeak_row is None:
|
||||
return None
|
||||
squeak = CSqueak(
|
||||
nVersion=squeak_row['nVersion'],
|
||||
hashEncContent=squeak_row['hashEncContent'],
|
||||
hashReplySqk=squeak_row['hashReplySqk'],
|
||||
hashBlock=squeak_row['hashBlock'],
|
||||
nBlockHeight=squeak_row['nBlockHeight'],
|
||||
scriptPubKey=CScript(squeak_row['scriptPubKey']),
|
||||
hashDataKey=squeak_row['hashDataKey'],
|
||||
vchIv=squeak_row['vchIv'],
|
||||
nTime=squeak_row['nTime'],
|
||||
nNonce=squeak_row['nNonce'],
|
||||
encContent=CSqueakEncContent(squeak_row['encContent']),
|
||||
scriptSig=CScript(squeak_row['scriptSig']),
|
||||
vchDataKey=squeak_row['vchDataKey'],
|
||||
)
|
||||
CheckSqueak(squeak)
|
||||
return squeak
|
||||
return self._row_to_squeak(squeak_row)
|
||||
|
||||
def delete_squeak(self):
|
||||
pass
|
||||
|
||||
def unlock_squeak(self):
|
||||
pass
|
||||
|
||||
def get_squeaks_to_upload(self, address, min_block=0):
|
||||
"""Get all squeaks that need to be uploaded.
|
||||
"""
|
||||
with self.db_factory.make_conn() as conn:
|
||||
squeak_row = (
|
||||
conn
|
||||
.execute(
|
||||
"SELECT s.hash, nVersion, hashEncContent, hashReplySqk, hashBlock, nBlockHeight, scriptPubKey, hashDataKey, vchIv, nTime, nNonce, encContent, scriptSig, vchDataKey"
|
||||
" FROM squeak s"
|
||||
" WHERE s.address = ?",
|
||||
(address,),
|
||||
)
|
||||
.fetchone()
|
||||
)
|
||||
if squeak_rows is None:
|
||||
return None
|
||||
squeaks = []
|
||||
for squeak_row in squeak_rows:
|
||||
squeak = self._row_to_squeak(squeak_row)
|
||||
squeaks.append(squeak)
|
||||
return squeaks
|
||||
|
||||
def mark_squeak_uploaded(self, squeak_hash):
|
||||
"""Mark the given squeak as one that needs to be uploaded.
|
||||
"""
|
||||
with self.db_factory.make_conn() as conn:
|
||||
conn.execute(
|
||||
"INSERT INTO upload (squeakHash, complete) VALUES (?, ?)",
|
||||
(squeak_hash, 1),
|
||||
)
|
||||
|
||||
def is_squeak_uploaded(self, squeak_hash):
|
||||
"""True if the squeak has already been uploaded.
|
||||
"""
|
||||
with self.db_factory.make_conn() as conn:
|
||||
upload_row = (
|
||||
conn
|
||||
.execute(
|
||||
"SELECT u.squeakHash, complete"
|
||||
" FROM upload u"
|
||||
" WHERE u.squeakHash = ?",
|
||||
(squeak_hash,),
|
||||
)
|
||||
.fetchone()
|
||||
)
|
||||
complete = upload_row['complete']
|
||||
return complete == 1
|
||||
|
|
|
|||
56
squeaknode/client/uploader.py
Normal file
56
squeaknode/client/uploader.py
Normal file
|
|
@ -0,0 +1,56 @@
|
|||
import logging
|
||||
import threading
|
||||
|
||||
from squeak.core.signing import CSigningKey
|
||||
from squeak.core.signing import CSqueakAddress
|
||||
|
||||
from squeaknode.client.squeak_store import SqueakStore
|
||||
from squeaknode.common.blockchain_client import BlockchainClient
|
||||
from squeaknode.common.lightning_client import LightningClient
|
||||
from squeaknode.common.squeak_maker import SqueakMaker
|
||||
from squeaknode.client.db import SQLiteDBFactory
|
||||
from squeaknode.client.hub_store import HubStore
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Uploader(object):
|
||||
"""Uploads squeaks to the hubs.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hub_store: HubStore,
|
||||
squeak_store: SqueakStore,
|
||||
address: CSqueakAddress,
|
||||
stopped: threading.Event,
|
||||
) -> None:
|
||||
self.hub_store = hub_store
|
||||
self.squeak_store = squeak_store
|
||||
self.address = address
|
||||
self.stopped = stopped
|
||||
|
||||
def upload_squeak(self, squeak):
|
||||
squeak_hash = squeak.GetHash()
|
||||
if self.squeak_store.is_squeak_uploaded(squeak_hash):
|
||||
# TODO: Upload squeak here.
|
||||
##
|
||||
logger.info("Upload the squeak here.")
|
||||
self.squeak_store.mark_squeak_uploaded(squeak_hash)
|
||||
|
||||
def upload_squeaks(self):
|
||||
for squeak in self.squeak_store.get_squeaks_to_upload(self.address):
|
||||
self.upload_squeak(squeak)
|
||||
|
||||
def start(self):
|
||||
while True:
|
||||
try:
|
||||
self.upload_squeaks()
|
||||
except:
|
||||
self.stopped.set()
|
||||
|
||||
# Check if client is stopped.
|
||||
is_stopped = self.stopped.wait(10)
|
||||
if is_stopped:
|
||||
break
|
||||
Loading…
Add table
Add a link
Reference in a new issue