atomic_p2p.abc package

Submodules

atomic_p2p.abc.authenticator module

class atomic_p2p.abc.authenticator.AuthenticatorABC

Bases: abc.ABC

Authenticator Absctract Class The absctract class for implementation.

_authenticate_packet(pkt, **kwargs)

The function for authenticate a packet. Actual examine logic is in this function, and after if-statement will call _on_pass method or _on_fail method to make flow control.

Return type

None

_on_fail(sock, pkt, **kwargs)

The method will be execute when authentication failed.

Return type

None

_on_pass(sock, pkt, **kwargs)

The method will be execute when authentication passed.

Return type

None

atomic_p2p.abc.topology module

class atomic_p2p.abc.topology.TopologyABC

Bases: abc.ABC

Topology Absctract Class The absctract class for implementation.

This abstract class should cowork with HandleableMixin and AuthenticatorABC.

add_peer_in_net(peer_info, **kwargs)

Return type

None

del_peer_in_net(peer_info, **kwargs)

Return type

bool

get_peer_info_by_conn(conn, **kwargs)

The function get peer info by given socket.

Return type

Union[None, _Forwardref]

get_peer_info_by_host(host, **kwargs)

The function get peer info by given host.

Return type

Union[None, _Forwardref]

handler_broadcast_packet(host, pkt_type, **kwargs)

Return type

None

handler_unicast_packet(host, pkt_type, **kwargs)

Return type

None

is_peer_in_net(info, **kwargs)

Return type

bool

join_net(host, **kwargs)

Return type

None

join_net_by_DNS(domain, ns=None, **kwargs)

Return type

None

leave_net(**kwargs)

Return type

None

on_packet_to_route(sock, pkt, **kwargs)

Call by AuthenticatorABC after a packet passed the authentication. The content should be like:

handler = self.select_handler(pkt_type=pkt._type) if handler is None:

self.logger.info(“Unknown packet type: {}”.format(pkt._type))

else:

return handler.on_recv(src=pkt.src, pkt=pkt, sock=sock)

Notice that select_handler method is from HandleableMixin.

topology_register_handler()

Call by a peer class to register essentials handlers. The content should be like:

installing_handlers = [

… AnyHandlerYouNeed, …

] for each in installing_handlers:

self.register_handler(handler=each)

Notice that register_handler method is from HandleableMixin & each element inside installing_handlers should be a Handler object.

Module contents

class atomic_p2p.abc.AuthenticatorABC

Bases: abc.ABC

Authenticator Absctract Class The absctract class for implementation.

_authenticate_packet(pkt, **kwargs)

The function for authenticate a packet. Actual examine logic is in this function, and after if-statement will call _on_pass method or _on_fail method to make flow control.

Return type

None

_on_fail(sock, pkt, **kwargs)

The method will be execute when authentication failed.

Return type

None

_on_pass(sock, pkt, **kwargs)

The method will be execute when authentication passed.

Return type

None

class atomic_p2p.abc.TopologyABC

Bases: abc.ABC

Topology Absctract Class The absctract class for implementation.

This abstract class should cowork with HandleableMixin and AuthenticatorABC.

add_peer_in_net(peer_info, **kwargs)

Return type

None

del_peer_in_net(peer_info, **kwargs)

Return type

bool

get_peer_info_by_conn(conn, **kwargs)

The function get peer info by given socket.

Return type

Union[None, _Forwardref]

get_peer_info_by_host(host, **kwargs)

The function get peer info by given host.

Return type

Union[None, _Forwardref]

handler_broadcast_packet(host, pkt_type, **kwargs)

Return type

None

handler_unicast_packet(host, pkt_type, **kwargs)

Return type

None

is_peer_in_net(info, **kwargs)

Return type

bool

join_net(host, **kwargs)

Return type

None

join_net_by_DNS(domain, ns=None, **kwargs)

Return type

None

leave_net(**kwargs)

Return type

None

on_packet_to_route(sock, pkt, **kwargs)

Call by AuthenticatorABC after a packet passed the authentication. The content should be like:

handler = self.select_handler(pkt_type=pkt._type) if handler is None:

self.logger.info(“Unknown packet type: {}”.format(pkt._type))

else:

return handler.on_recv(src=pkt.src, pkt=pkt, sock=sock)

Notice that select_handler method is from HandleableMixin.

topology_register_handler()

Call by a peer class to register essentials handlers. The content should be like:

installing_handlers = [

… AnyHandlerYouNeed, …

] for each in installing_handlers:

self.register_handler(handler=each)

Notice that register_handler method is from HandleableMixin & each element inside installing_handlers should be a Handler object.