chainlib-eth/chainlib/eth/nonce.py

# external imports
from hexathon import (
        add_0x,
        strip_0x,
        )

# local imports
from chainlib.nonce import NonceOracle as BaseNonceOracle
from chainlib.jsonrpc import JSONRPCRequest


def nonce(address, confirmed=False, id_generator=None):
    """Generate json-rpc query to retrieve next nonce of address from node.

    :param address: Address to retrieve nonce for, in hex
    :type address: str
    :param id_generator: json-rpc id generator 
    :type id_generator: chainlib.connection.JSONRPCIdGenerator
    :rtype: dict
    :returns: rpc query object
    """
    j = JSONRPCRequest(id_generator)
    o = j.template()
    o['method'] = 'eth_getTransactionCount'
    o['params'].append(address)
    if confirmed:
        o['params'].append('latest')
    else:
        o['params'].append('pending')
    return j.finalize(o)


def nonce_confirmed(address, id_generator=None):
    return nonce(address, confirmed=True, id_generator=id_generator)


class NonceOracle(BaseNonceOracle):
    """Base class for the nonce parameter helpers.

    :param address: Address to retireve nonce for, in hex
    :type address: str
    :param id_generator: json-rpc id generator 
    :type id_generator: chainlib.connection.JSONRPCIdGenerator
    """
    def __init__(self, address, id_generator=None):
        self.id_generator = id_generator
        super(NonceOracle, self).__init__(add_0x(address))


    def get_nonce(self):
        """Load initial nonce value.
        """
        raise NotImplementedError('Class must be extended')


    def next_nonce(self):
        """Return next nonce value and advance.

        :rtype: int
        :returns: Next nonce for address.
        """
        n = self.nonce
        self.nonce += 1
        return n


class RPCNonceOracle(NonceOracle):
    """JSON-RPC only nonce parameter helper.

    :param address: Address to retireve nonce for, in hex
    :type address: str
    :param conn: RPC connection
    :type conn: chainlib.connection.RPCConnection
    :param id_generator: json-rpc id generator 
    :type id_generator: chainlib.connection.JSONRPCIdGenerator
    """
    def __init__(self, address, conn, id_generator=None):
        self.conn = conn
        super(RPCNonceOracle, self).__init__(address, id_generator=id_generator)


    def get_nonce(self):
        """Load and return nonce value from network.

        Note! First call to next_nonce after calling get_nonce will return the same value!

        :rtype: int
        :returns: Initial nonce
        """
        o = nonce(self.address, id_generator=self.id_generator)
        r = self.conn.do(o)
        n = strip_0x(r)
        return int(n, 16)


class OverrideNonceOracle(NonceOracle):
    """Manually set initial nonce value.

    :param address: Address to retireve nonce for, in hex
    :type address: str
    :param nonce: Nonce value
    :type nonce: int
    :param id_generator: json-rpc id generator (not used)
    :type id_generator: chainlib.connection.JSONRPCIdGenerator
    """
    def __init__(self, address, nonce, id_generator=None):
        self.initial_nonce = nonce
        self.nonce = self.initial_nonce
        super(OverrideNonceOracle, self).__init__(address, id_generator=id_generator)


    def get_nonce(self):
        """Returns initial nonce value set at object construction.

        :rtype: int
        :returns: Initial nonce value.
        """
        return self.initial_nonce


DefaultNonceOracle = RPCNonceOracle