import asyncio
import sys
import json
import logging
from bgphoria import BGPClient
from bgphoria.constants import ADD_PATH
def configure_default_logger():
if not logging.getLogger().hasHandlers():
logging.basicConfig(
level=logging.INFO, # Set default logging level
format="%(asctime)s - %(levelname)s - %(message)s",
stream=sys.stdout # Log to stdout
)
def update_test_cb(update):
print(json.dumps(update, indent=4))
if __name__ == "__main__":
PEER_IP = "192.168.1.2" # Replace with your BGP peer's IP
LOCAL_AS = 123456 # Replace with your AS number
LOCAL_BGP_ID = "192.168.1.1" # Replace with your BGP identifier
# Configure module-level logger
configure_default_logger()
log = logging.getLogger(__name__)
try:
print('starting')
client = BGPClient("PEER_ID",
PEER_IP,
LOCAL_AS,
LOCAL_AS,
LOCAL_BGP_ID,
update_callback=update_test_cb,
ipv4_unicast=True,
ipv4_unicast_add_path=ADD_PATH.RECEIVE_ONLY)
loop = asyncio.get_event_loop()
loop.create_task(client.connect()) # Connect Event
loop.run_forever() # Keeps the loop running indefinitely
except KeyboardInterrupt:
print('stopping')