Compare commits

...

8 Commits

5 changed files with 64 additions and 6 deletions

View File

@ -121,20 +121,25 @@ broker = config.get('CELERY_BROKER_URL')
if broker[:4] == 'file': if broker[:4] == 'file':
bq = tempfile.mkdtemp() bq = tempfile.mkdtemp()
bp = tempfile.mkdtemp() bp = tempfile.mkdtemp()
current_app.conf.update({ conf_update = {
'broker_url': broker, 'broker_url': broker,
'broker_transport_options': { 'broker_transport_options': {
'data_folder_in': bq, 'data_folder_in': bq,
'data_folder_out': bq, 'data_folder_out': bq,
'data_folder_processed': bp, 'data_folder_processed': bp,
}, },
}, }
) if config.true('CELERY_DEBUG'):
conf_update['result_extended'] = True
current_app.conf.update(conf_update)
logg.warning('celery broker dirs queue i/o {} processed {}, will NOT be deleted on shutdown'.format(bq, bp)) logg.warning('celery broker dirs queue i/o {} processed {}, will NOT be deleted on shutdown'.format(bq, bp))
else: else:
current_app.conf.update({ conf_update = {
'broker_url': broker, 'broker_url': broker,
}) }
if config.true('CELERY_DEBUG'):
conf_update['result_extended'] = True
current_app.conf.update(conf_update)
result = config.get('CELERY_RESULT_URL') result = config.get('CELERY_RESULT_URL')
if result[:4] == 'file': if result[:4] == 'file':

View File

@ -1,3 +1,4 @@
[celery] [celery]
broker_url = redis:// broker_url = redis://
result_url = redis:// result_url = redis://
debug = 0

View File

@ -1,3 +1,4 @@
[celery] [celery]
broker_url = redis://localhost:63379 broker_url = redis://localhost:63379
result_url = redis://localhost:63379 result_url = redis://localhost:63379
debug = 0

View File

@ -0,0 +1,50 @@
# standard imports
import sys
import os
import logging
import json
# external imports
import celery
import confini
# local imports
from cic_eth.api import Api
logging.basicConfig(level=logging.DEBUG)
logg = logging.getLogger()
script_dir = os.path.realpath(os.path.dirname(__file__))
config_dir = os.path.join(script_dir, '..', 'config')
config = confini.Config(config_dir, os.environ.get('CONFINI_ENV_PREFIX'))
config.process()
celery_app = celery.Celery(broker=config.get('CELERY_BROKER_URL'), backend=config.get('CELERY_RESULT_URL'), result_extended=True)
class Fmtr(celery.utils.graph.GraphFormatter):
def label(self, obj):
super(Fmtr, self).label(obj)
if obj != None:
if obj.name == None:
raise RuntimeError('task name is not defined. Did you run celery with result_extended=True?')
return obj.name
def main():
api = Api(
config.get('CIC_CHAIN_SPEC'),
queue='cic-eth',
#callback_param='{}:{}:{}:{}'.format(args.redis_host_callback, args.redis_port_callback, redis_db, redis_channel),
#callback_task='cic_eth.callbacks.redis.redis',
#callback_queue=args.q,
)
t = api.create_account(register=False)
t.get_leaf()
t.build_graph(intermediate=True, formatter=Fmtr()).to_dot(sys.stdout)
if __name__ == '__main__':
main()

View File

@ -248,6 +248,7 @@ services:
BANCOR_DIR: ${BANCOR_DIR:-/usr/local/share/cic/bancor} BANCOR_DIR: ${BANCOR_DIR:-/usr/local/share/cic/bancor}
CELERY_BROKER_URL: ${CELERY_BROKER_URL:-redis://redis} CELERY_BROKER_URL: ${CELERY_BROKER_URL:-redis://redis}
CELERY_RESULT_URL: ${CELERY_RESULT_URL:-redis://redis} CELERY_RESULT_URL: ${CELERY_RESULT_URL:-redis://redis}
CELERY_DEBUG: ${CELERY_DEBUG:-1}
SIGNER_SOCKET_PATH: ${SIGNER_SOCKET_PATH:-ipc:///run/crypto-dev-signer/jsonrpc.ipc} SIGNER_SOCKET_PATH: ${SIGNER_SOCKET_PATH:-ipc:///run/crypto-dev-signer/jsonrpc.ipc}
SIGNER_SECRET: ${SIGNER_SECRET:-deadbeef} SIGNER_SECRET: ${SIGNER_SECRET:-deadbeef}
ETH_ACCOUNT_ACCOUNTS_INDEX_WRITER: ${DEV_ETH_ACCOUNT_ACCOUNTS_INDEX_WRITER:-0xACB0BC74E1686D62dE7DC6414C999EA60C09F0eA} ETH_ACCOUNT_ACCOUNTS_INDEX_WRITER: ${DEV_ETH_ACCOUNT_ACCOUNTS_INDEX_WRITER:-0xACB0BC74E1686D62dE7DC6414C999EA60C09F0eA}