chaind-eth

Queue server for ethereum
Info | Log | Files | Refs | README | LICENSE

send.py (4202B)

      1 # standard imports
      2 import os
      3 import logging
      4 import sys
      5 import datetime
      6 import enum
      7 import re
      8 import stat
      9 import socket
     10 
     11 # external imports
     12 import chainlib.eth.cli
     13 from chaind import Environment
     14 from chainlib.eth.gas import price
     15 from chainlib.chain import ChainSpec
     16 from hexathon import strip_0x
     17 
     18 # local imports
     19 from chaind_eth.cli.process import Processor
     20 from chaind_eth.cli.csv import CSVProcessor
     21 from chaind.error import TxSourceError
     22 from chaind_eth.cli.resolver import (
     23         DefaultResolver,
     24         LookNoop,
     25         TokenIndexLookup,
     26         )
     27 
     28 logging.basicConfig(level=logging.WARNING)
     29 logg = logging.getLogger()
     30 
     31 script_dir = os.path.dirname(os.path.realpath(__file__)) 
     32 config_dir = os.path.join(script_dir, '..', 'data', 'config')
     33 
     34 
     35 arg_flags = chainlib.eth.cli.argflag_std_write
     36 argparser = chainlib.eth.cli.ArgumentParser(arg_flags)
     37 argparser.add_argument('--socket', dest='socket', type=str, help='Socket to send transactions to')
     38 argparser.add_argument('--token-index', dest='token_index', type=str, help='Token resolver index')
     39 argparser.add_positional('source', required=False, type=str, help='Transaction source file')
     40 args = argparser.parse_args()
     41 
     42 extra_args = {
     43         'socket': None,
     44         'source': None,
     45         'token_index': None,
     46         }
     47 
     48 env = Environment(domain='eth', env=os.environ)
     49 config = chainlib.eth.cli.Config.from_args(args, arg_flags, extra_args=extra_args, base_config_dir=config_dir)
     50 
     51 wallet = chainlib.eth.cli.Wallet()
     52 wallet.from_config(config)
     53 
     54 rpc = chainlib.eth.cli.Rpc(wallet=wallet)
     55 conn = rpc.connect_by_config(config)
     56 
     57 chain_spec = ChainSpec.from_chain_str(config.get('CHAIN_SPEC'))
     58 
     59 class OpMode(enum.Enum):
     60     STDOUT = 'standard_output'
     61     UNIX = 'unix_socket'
     62 mode = OpMode.STDOUT
     63 
     64 re_unix = r'^ipc://(/.+)'
     65 m = re.match(re_unix, config.get('_SOCKET', ''))
     66 if m != None:
     67     config.add(m.group(1), '_SOCKET', exists_ok=True)
     68     r = 0
     69     try:
     70         stat_info = os.stat(config.get('_SOCKET'))
     71         if not stat.S_ISSOCK(stat_info.st_mode):
     72             r = 1
     73     except FileNotFoundError:
     74         r = 1
     75 
     76     if r > 0:
     77         sys.stderr.write('{} is not a socket\n'.format(config.get('_SOCKET')))
     78         sys.exit(1)
     79     
     80     mode = OpMode.UNIX
     81 
     82 logg.info('using mode {}'.format(mode.value))
     83 
     84 if config.get('_SOURCE') == None:
     85     sys.stderr.write('source data missing\n')
     86     sys.exit(1)
     87 
     88 
     89 class Outputter:
     90 
     91     def __init__(self, mode):
     92         self.out = getattr(self, 'do_' + mode.value)
     93 
     94 
     95     def do(self, hx):
     96         return self.out(hx)
     97 
     98 
     99     def do_standard_output(self, hx):
    100         #sys.stdout.write(hx + '\n')
    101         return hx
    102 
    103 
    104     def do_unix_socket(self, hx):
    105         s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    106         s.connect(config.get('_SOCKET'))
    107         s.send(hx.encode('utf-8'))
    108         r = s.recv(64+4)
    109         logg.debug('r {}'.format(r))
    110         s.close()
    111         return r[4:].decode('utf-8')
    112 
    113 
    114 def main():
    115     signer = rpc.get_signer()
    116 
    117 
    118     # TODO: make resolvers pluggable
    119     token_resolver = DefaultResolver(chain_spec, conn, sender_address=rpc.get_sender_address())
    120 
    121     noop_lookup = LookNoop(check=not config.true('_UNSAFE'))
    122     token_resolver.add_lookup(noop_lookup, 'noop')
    123   
    124     if config.get('_TOKEN_INDEX') != None:
    125         token_index_lookup = TokenIndexLookup(chain_spec, signer, rpc.get_gas_oracle(), rpc.get_nonce_oracle(), config.get('_TOKEN_INDEX'))
    126         token_resolver.add_lookup(token_index_lookup, reverse=config.get('_TOKEN_INDEX'))
    127     
    128     processor = Processor(wallet.get_signer_address(), wallet.get_signer(), config.get('_SOURCE'), chain_spec, rpc.get_gas_oracle(), rpc.get_nonce_oracle(), resolver=token_resolver)
    129     processor.add_processor(CSVProcessor())
    130 
    131     sends = None
    132     try:
    133         sends = processor.load()
    134     except TxSourceError as e:
    135         sys.stderr.write('processing error: {}. processors: {}\n'.format(str(e), str(processor)))
    136         sys.exit(1)
    137 
    138 
    139     tx_iter = iter(processor)
    140     out = Outputter(mode)
    141     while True:
    142         tx = None
    143         try:
    144             tx_bytes = next(tx_iter)
    145         except StopIteration:
    146             break
    147         tx_hex = tx_bytes.hex()
    148         print(out.do(tx_hex))
    149 
    150 
    151 if __name__ == '__main__':
    152     main()