@@ -99,6 +99,7 @@ def __init__(
9999 smooth : bool ,
100100 verify_and_reencode : bool ,
101101 timeout : Optional [float ],
102+ keepalive_interval_seconds : Optional [float ] = None ,
102103 ) -> None :
103104 self ._resource = resource
104105 self ._token_id = token_id
@@ -108,12 +109,22 @@ def __init__(
108109 self ._smooth = smooth
109110 self ._verify_and_reencode = verify_and_reencode
110111 self ._timeout = timeout
112+ self ._keepalive_interval = keepalive_interval_seconds
111113 self ._slots = [_StreamPoolSlot () for _ in range (connections )]
112114 self ._idle = deque (self ._slots )
113115 self ._idle_lock = threading .Lock ()
114116 self ._idle_ready = threading .Condition (self ._idle_lock )
115117 self ._state_lock = threading .Lock ()
116118 self ._closed = False
119+ self ._keepalive_stop = threading .Event ()
120+ self ._keepalive_thread : Optional [threading .Thread ] = None
121+ if keepalive_interval_seconds is not None :
122+ self ._keepalive_thread = threading .Thread (
123+ target = self ._keepalive_loop ,
124+ name = "ofspectrum-stream-keepalive" ,
125+ daemon = True ,
126+ )
127+ self ._keepalive_thread .start ()
117128
118129 def __enter__ (self ) -> "StreamEncodePool" :
119130 return self
@@ -132,6 +143,10 @@ def close(self) -> None:
132143 if self ._closed :
133144 return
134145 self ._closed = True
146+ self ._keepalive_stop .set ()
147+ thread = self ._keepalive_thread
148+ if thread is not None and thread is not threading .current_thread ():
149+ thread .join (timeout = 1.0 )
135150 with self ._idle_ready :
136151 self ._idle_ready .notify_all ()
137152
@@ -166,6 +181,77 @@ def close(self) -> None:
166181 finally :
167182 slot .lock .release ()
168183
184+ def heartbeat (self , * , timeout : Optional [float ] = None ) -> bool :
185+ """Refresh every pooled stream connection without encoding audio.
186+
187+ Busy slots are skipped because an in-flight encode already keeps them
188+ alive. Dead connections are replaced silently. This method never raises;
189+ a ``False`` result means no slot confirmed an ack, but later ``encode()``
190+ still uses the existing fail-open reconnect path.
191+ """
192+ if self .closed :
193+ return False
194+ deadline = None
195+ if timeout is not None :
196+ deadline = time .monotonic () + float (timeout )
197+ elif self ._timeout is not None :
198+ deadline = time .monotonic () + float (self ._timeout )
199+ confirmed = False
200+ for slot in self ._slots :
201+ if self ._heartbeat_slot (slot , deadline ):
202+ confirmed = True
203+ return confirmed
204+
205+ def _keepalive_loop (self ) -> None :
206+ interval = self ._keepalive_interval
207+ if interval is None :
208+ return
209+ while not self ._keepalive_stop .wait (timeout = interval ):
210+ if self .closed :
211+ return
212+ self .heartbeat (timeout = self ._CONNECT_TIMEOUT_SECONDS )
213+
214+ def _heartbeat_slot (
215+ self ,
216+ slot : _StreamPoolSlot ,
217+ deadline : Optional [float ],
218+ ) -> bool :
219+ if not slot .lock .acquire (blocking = False ):
220+ return True
221+ try :
222+ if self .closed :
223+ return False
224+ try :
225+ return self ._heartbeat_once (slot , deadline )
226+ except Exception :
227+ self ._discard_connection (slot )
228+ try :
229+ return self ._heartbeat_once (slot , deadline )
230+ except Exception :
231+ self ._discard_connection (slot )
232+ return False
233+ finally :
234+ slot .lock .release ()
235+
236+ def _heartbeat_once (
237+ self ,
238+ slot : _StreamPoolSlot ,
239+ deadline : Optional [float ],
240+ ) -> bool :
241+ self ._ensure_connection (slot , deadline )
242+ websocket = slot .websocket
243+ if websocket is None :
244+ return False
245+ websocket .send (json .dumps ({"type" : "heartbeat" }))
246+ event = self ._control_event (self ._recv (websocket , deadline ))
247+ if event .get ("type" ) != "heartbeat_ack" :
248+ raise OfSpectrumError (
249+ message = "Streaming pool received an invalid heartbeat response" ,
250+ code = "StreamingPoolProtocol" ,
251+ )
252+ slot .last_ok_monotonic = time .monotonic ()
253+ return True
254+
169255 def encode (
170256 self ,
171257 audio : Union [str , Path , BinaryIO , bytes ],
@@ -754,12 +840,19 @@ def open_stream_pool(
754840 smooth : bool = True ,
755841 verify_and_reencode : bool = True ,
756842 timeout : Optional [float ] = None ,
843+ keepalive_interval_seconds : Optional [float ] = None ,
757844 ) -> StreamEncodePool :
758845 """Create a thread-safe persistent pool for repeated stream encodes.
759846
760847 Each connection has a stable opaque slot identifier. The pool config is
761848 fixed at creation time; ``encode()`` accepts file paths, file objects,
762849 or bytes with the same number of audio channels as the pool.
850+
851+ ``keepalive_interval_seconds`` sends ``{"type": "heartbeat"}`` on every
852+ pooled connection so Neo does not idle-retire the model session. Use a
853+ value below the server idle window (default 300s), typically 120. Opening
854+ the pool and completing ``admitted`` / ``ready`` warms the connection;
855+ do not send dummy audio just to warm up. ``None`` disables keepalive.
763856 """
764857 if not token_id :
765858 raise ValueError ("token_id is required" )
@@ -783,6 +876,12 @@ def open_stream_pool(
783876 isinstance (timeout , bool ) or not math .isfinite (timeout ) or timeout <= 0.0
784877 ):
785878 raise ValueError ("timeout must be a finite positive number or None" )
879+ if keepalive_interval_seconds is not None and (
880+ isinstance (keepalive_interval_seconds , bool )
881+ or not math .isfinite (keepalive_interval_seconds )
882+ or keepalive_interval_seconds <= 0.0
883+ ):
884+ raise ValueError ("keepalive_interval_seconds must be a positive number or None" )
786885
787886 pool = StreamEncodePool (
788887 self ,
@@ -794,6 +893,7 @@ def open_stream_pool(
794893 smooth = smooth ,
795894 verify_and_reencode = verify_and_reencode ,
796895 timeout = timeout ,
896+ keepalive_interval_seconds = keepalive_interval_seconds ,
797897 )
798898 with self ._stream_pools_lock :
799899 self ._stream_pools .add (pool )
0 commit comments