The [[Node|Cellframe Node]] Python extensions provide a set of classes and functions for working with the [[Mempool|mempool]]. Each mempool belongs to a [[Chain|chain]], which in turn belongs to a [[Network|network]]. It is possible to retrieve [[Data element (Datum)|datums]] from the mempool using the `Mempool.list()` method. Since the mempool is part of the [[Global Database (GDB)]], all data is stored there as bytes. You can decode this data using the `Mempool.datumExtract()` method. You can also register a handler function for adding or deleting data in the mempool.
The code below shows how to:
1. Create handler functions for mempool notifications.
2. Retrieve datums from the mempool.
3. Deserialize datums from the mempool.
4. Register notification handlers for mempool changes.
5. Add datums to the mempool.
```python
from CellFrame.Network import Net
from DAP.Core import logIt
from CellFrame.Consensus import Block
from CellFrame.Chain import Mempool
def mempool_notificator(op_code, group, key, value, *other):
"""
Handler for mempool notifications.
Args:
op_code (str): Operation code indicating the type of change.
group (str): The group where the changes occurred.
key (str): The key associated with the change.
value (bytes): The value associated with the change.
*other: Additional arguments.
"""
# Decode the datum from the value using the datumExtract() static method
datum = Mempool.datumExtract(value)
logIt.notice("Mempool callback with parameters:")
logIt.notice(f"Operation code: {op_code}")
logIt.notice(f"Group alias: {group}")
logIt.notice(f"Key: {key}")
logIt.notice(f"Datum hash: {datum.hash}")
logIt.notice(f"Additional arguments: {other}")
return
def data_from_mempool(net, chain) -> list[bytes] | None:
"""
Retrieve datums from the mempool and log their hashes.
Args:
net (Net): The network object.
chain (Chain): The chain object.
Returns:
list[bytes] | None: A list of serialized datum objects from the mempool.
"""
datums = Mempool.list(net, chain) # dict[str, Datum]
if datums:
num_items = len(datums)
logIt.notice(f"Number of records in mempool: {num_items}")
for key, value in datums.items():
logIt.notice(f"{key=}, {value=}")
return list(datums.values())
else:
logIt.notice("Mempool is empty.")
return None
def retrieve_some_datum(chain):
"""
Retrieve the last datum from the chain.
Args:
chain (Chain): The chain object.
Returns:
Datum: The last datum in the chain.
"""
# Select the last atom in the chain:
atom = chain.getAtoms(1, 1, False)
if atom:
# Since
logIt.notice(f"{Block(atom).hash}") # ОТЛАДКА
# Retrieve the last datum in the atom:
datum = chain.atomGetDatums(atom)[-1]
else:
logIt.error("Unable to retrieve datum.")
return datum
def init():
net_name = "KelVPN"
logIt.notice("START TEST_1")
KELVPN_NET = Net.byName(net_name)
main_chain = KELVPN_NET.getChainByName("main")
main_chain.addMempoolNotify(mempool_notificator, ())
datums = data_from_mempool(KELVPN_NET, main_chain)
if datums:
# Get the last datum from the mempool
last_datum = datums[-1]
# Serialize the datum using the "raw" property
serialized_datum = last_datum.raw
deserialized_datum = Mempool.datumExtract(serialized_datum)
# Compare the hashes to ensure successful extraction
if last_datum.hash == deserialized_datum.hash:
logIt.notice(f"The datum with hash {last_datum.hash} has been successfully extracted from the mempool")
else:
logIt.notice("Something went wrong: the hash of the datum before serialization and after extraction from the mempool do not match.")
# Add some atom to the mempool
some_datum = retrieve_some_datum(main_chain)
if some_datum:
logIt.notice(f"{some_datum.hash=}")
result = Mempool.addDatum(main_chain, some_datum)
if result:
logIt.notice(f"Datum with hash {result} added to the mempool")
else:
logIt.notice("Something went wrong while adding the datum to the mempool.")
return 0
```