Skip to content

plugins.aea-cli-benchmark.aea_cli_benchmark.case_tx_generate.case

Ledger TX generation and processing benchmark.

LedgerApiHandler Objects

class LedgerApiHandler(Handler)

Dummy handler to handle messages.

setup

def setup() -> None

Noop setup.

teardown

def teardown() -> None

Noop teardown.

handle

def handle(ledger_api_msg: Message) -> None

Handle incoming message.

GenericSigningHandler Objects

class GenericSigningHandler(Handler)

Implement the signing handler.

setup

def setup() -> None

Implement the setup for the handler.

handle

def handle(message: Message) -> None

Implement the reaction to a message.

Arguments:

  • message: the message

teardown

def teardown() -> None

Implement the handler teardown.

Case Objects

class Case()

TBenchmark case implementation.

__init__

def __init__(ledger_id: str, ledger_api_config: Dict, private_keys: Dict)

Init case.

Arguments:

  • ledger_id: str, ledger id, one of fetchai, ethereum
  • ledger_api_config: config for ledger connection
  • private_keys: private keys dict to use for wallet contruction

ledger_handler

@property
def ledger_handler() -> LedgerApiHandler

Get ledger api handler instance.

tx_settled_counter

@property
def tx_settled_counter() -> int

Get amount of txs settled.

wait_tx_settled

def wait_tx_settled() -> None

Wait for tx settled.

ledger_api_dialogues

@property
def ledger_api_dialogues() -> LedgerApiDialogues

Get ledger api dialogues.

my_addr

@property
def my_addr() -> str

Get my agent address.

make_ledger_msg

def make_ledger_msg(sender_address: str,
                    counterparty_address: str) -> LedgerApiMessage

Make ledger api message to be signed and published over ledger netework.

start_agent

def start_agent() -> None

Construct and start agent.

stop_agent

def stop_agent() -> None

Stop agent.

put_message_and_wait

def put_message_and_wait(msg: Message) -> None

Put ledger api message and wait tx constructed, signed and settled.

run

def run(time_in_seconds: float) -> Tuple[int, float]

Run a test case.

run

def run(ledger_id: str,
        running_time: float) -> List[Tuple[str, Union[int, float]]]

Check tx processing speed.