popov/gogo2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

handle ws exceptions on rpc calls

Browse Source
This commit is contained in:
Dobromir Popov 2024-10-03 01:48:11 +03:00
parent b9405424ea
commit 3db80762a0
1 changed files with 58 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

87
crypto/sol/app.py
View File

@ -40,9 +40,6 @@ DISPLAY_CURRENCY = os.getenv('DISPLAY_CURRENCY', 'USD')
# Initialize Telegram Bot # Initialize Telegram Bot
bot = Bot(token=TELEGRAM_BOT_TOKEN) bot = Bot(token=TELEGRAM_BOT_TOKEN)
# Initialize logging
logging.basicConfig(level=logging.DEBUG)
# Token addresses (initialize with some known tokens) # Token addresses (initialize with some known tokens)
TOKEN_ADDRESSES = { TOKEN_ADDRESSES = {
"SOL": "So11111111111111111111111111111111111111112", "SOL": "So11111111111111111111111111111111111111112",
@ -462,37 +459,69 @@ async def on_logs(log):
async def subscribe_to_wallet(): async def subscribe_to_wallet():
uri = SOLANA_URL uri = SOLANA_URL
async with websockets.connect(uri) as websocket: reconnect_delay = 5 # Start with a 5-second delay
# Correct the `params` format to be an array max_reconnect_delay = 60 # Maximum delay of 60 seconds
request = {
"jsonrpc": "2.0", while True:
"id": 1, try:
"method": "logsSubscribe", async with websockets.connect(uri) as websocket:
"params": [ logger.info("Connected to Solana websocket")
{
"mentions": [FOLLOWED_WALLET] # Changed from YOUR_WALLET to FOLLOWED_WALLET request = {
}, "jsonrpc": "2.0",
{ "id": 1,
"commitment": "confirmed" "method": "logsSubscribe",
"params": [
{
"mentions": [FOLLOWED_WALLET]
},
{
"commitment": "confirmed"
}
]
} }
]
}
await websocket.send(json.dumps(request)) await websocket.send(json.dumps(request))
logger.info("Subscription request sent")
# Listen for messages while True:
while True: try:
response = await websocket.recv() response = await websocket.recv()
response_data = json.loads(response) response_data = json.loads(response)
if 'result' in response_data: if 'result' in response_data:
print(f"Subscription successful. Subscription id: {response_data['result']}") logger.info(f"Subscription successful. Subscription id: {response_data['result']}")
elif 'params' in response_data: elif 'params' in response_data:
await on_logs(response_data['params']['result']) await on_logs(response_data['params']['result'])
else: else:
print(f"Unexpected response: {response}") logger.warning(f"Unexpected response: {response}")
except websockets.exceptions.ConnectionClosedError as e:
logger.error(f"Connection closed unexpectedly: {e}")
break
except json.JSONDecodeError as e:
logger.error(f"Failed to decode JSON: {e}")
except Exception as e:
logger.error(f"An unexpected error occurred: {e}")
break
except websockets.exceptions.WebSocketException as e:
logger.error(f"WebSocket error: {e}")
except Exception as e:
logger.error(f"An unexpected error occurred: {e}")
logger.info(f"Attempting to reconnect in {reconnect_delay} seconds...")
await asyncio.sleep(reconnect_delay)
# Implement exponential backoff
reconnect_delay = min(reconnect_delay * 2, max_reconnect_delay)
logger = logging.getLogger(__name__)
async def main(): async def main():
# Initialize logging
logging.basicConfig(level=logging.DEBUG)
logging.basicConfig(level=logging.INFO) logging.basicConfig(level=logging.INFO)
await send_telegram_message("Solana Agent Application Started") await send_telegram_message("Solana Agent Application Started")
await list_initial_wallet_states() await list_initial_wallet_states()
await subscribe_to_wallet() await subscribe_to_wallet()