Fix: Prevent spurious retry attempts after subscription EndOfStream

[vjsingh1984]

Fixes #396 - Prevents spurious retry attempts after a subscription completes (receives EndOfStream).

Problem

After a subscription completes (e.g., OpenOrderEnd message), stray messages remaining in the message bus queue triggered unnecessary retry attempts. This caused:

• Warning spam: "retrying after unexpected response" logged up to 10 times
• Wasted CPU cycles: Up to 10 decode attempts per stray message
• Misleading logs: Users saw warnings even though subscription completed successfully

Solution

Track stream completion state with stream_ended: Arc<AtomicBool> field to prevent retries after EndOfStream:

pub struct Subscription<T> {
    stream_ended: Arc<AtomicBool>,  // Persists across async calls
    // ...
}

ProcessingResult::EndOfStream => {
    self.stream_ended.store(true, Ordering::Release);
    return None;
}
ProcessingResult::Retry => {
    // Stop retrying if stream has already ended
    if self.stream_ended.load(Ordering::Acquire) {
        return None;  // ✅ Blocks retries after completion
    }
    // ... normal retry logic
}

Key improvement: The stream_ended flag is stored as a struct field (not a local variable), so it persists across async next() calls. This prevents retries even when stray messages arrive in the shared message bus queue after the subscription has completed.

Changes

• Added stream_ended: Arc<AtomicBool> field to Subscription<T> struct
• Initialize to false in all constructors
• Update Clone implementation to include the field
• Set to true on EndOfStream, cancel(), and drop()
• Check the field before retrying on UnexpectedResponse

Testing

All existing tests pass. The fix is internal state management with no API changes.

Note

This fix addresses spurious retries for single subscription instances. For applications that create multiple subscriptions (e.g., polling open_orders() every 30 seconds), additional caching may be needed at the application layer to prevent creating duplicate subscriptions on the shared message bus.