Examples to Demonstrate KeepAlive and Failure for Not Using It

tests/edge_cases/keepalive/async/main.py

@@ -0,0 +1,33 @@
+# Copyright 2024 Deepgram SDK contributors. All Rights Reserved.
+# Use of this source code is governed by a MIT license that can be found in the LICENSE file.
+# SPDX-License-Identifier: MIT
+
+import asyncio
+import time
+import logging, verboselogs
+
+from deepgram import DeepgramClient, DeepgramClientOptions, LiveOptions
+
+
+async def main():
+    # for debugging
+    config: DeepgramClientOptions = DeepgramClientOptions(
+        verbose=logging.DEBUG, options={"keepalive": "true"}
+    )
+    deepgram: DeepgramClient = DeepgramClient("", config)
+
+    deepgram_connection = deepgram.listen.asynclive.v("1")
+
+    await deepgram_connection.start(LiveOptions())
+
+    # Wait for a while to simulate a long-running connection
+    await asyncio.sleep(600)
+
+    print("deadlock!")
+    try:
+        await deepgram_connection.finish()
+    finally:
+        print("no deadlock...")
+
+
+asyncio.run(main())

tests/edge_cases/keepalive/sync/main.py

@@ -0,0 +1,35 @@
+# Copyright 2024 Deepgram SDK contributors. All Rights Reserved.
+# Use of this source code is governed by a MIT license that can be found in the LICENSE file.
+# SPDX-License-Identifier: MIT
+
+import time
+import logging, verboselogs
+
+from deepgram import DeepgramClient, DeepgramClientOptions, LiveOptions
+
+
+def main():
+    # for debugging
+    config: DeepgramClientOptions = DeepgramClientOptions(
+        verbose=logging.DEBUG, options={"keepalive": "true"}
+    )
+    deepgram: DeepgramClient = DeepgramClient("", config)
+    # OR
+    # deepgram: DeepgramClient = DeepgramClient()
+
+    deepgram_connection = deepgram.listen.live.v("1")
+
+    deepgram_connection.start(LiveOptions())
+
+    # press any key to exit
+    input("\n\nPress Enter to exit...\n\n")
+
+    print("deadlock!")
+    try:
+        deepgram_connection.finish()
+    finally:
+        print("no deadlock...")
+
+
+if __name__ == "__main__":
+    main()

tests/expected_failures/async_live_timeout/main.py → tests/expected_failures/exercise_timeout/async/main.py

@@ -10,16 +10,18 @@
 
 
 async def main():
+    # for debugging
     config: DeepgramClientOptions = DeepgramClientOptions(verbose=logging.DEBUG)
     deepgram: DeepgramClient = DeepgramClient("", config)
+    # OR
+    # deepgram: DeepgramClient = DeepgramClient()
 
     deepgram_connection = deepgram.listen.asynclive.v("1")
 
     await deepgram_connection.start(LiveOptions())
 
-    time.sleep(
-        30
-    )  # Deepgram will close the connection after 10-15s of silence, followed with another 5 seconds for a ping
+    # Deepgram will close the connection after 10-15s of silence, followed with another 5 seconds for a ping
+    await asyncio.sleep(30)
 
     print("deadlock!")
     try:

tests/expected_failures/sync_live_timeout/main.py → tests/expected_failures/exercise_timeout/sync/main.py

@@ -9,16 +9,18 @@
 
 
 def main():
-    config: DeepgramClientOptions = DeepgramClientOptions(verbose=logging.DEBUG)
+    # for debugging
+    config: DeepgramClientOptions = DeepgramClientOptions(verbose=logging.SPAM)
     deepgram: DeepgramClient = DeepgramClient("", config)
+    # OR
+    # deepgram: DeepgramClient = DeepgramClient()
 
     deepgram_connection = deepgram.listen.live.v("1")
 
     deepgram_connection.start(LiveOptions())
 
-    time.sleep(
-        30
-    )  # Deepgram will close the connection after 10-15s of silence, followed with another 5 seconds for a ping
+    # Deepgram will close the connection after 10-15s of silence, followed with another 5 seconds for a ping
+    time.sleep(30)
 
     print("deadlock!")
     try: