Fix nats-pure 2.5.0 compatibility issue

Fixes connection establishment bug where stream_info config and state
attributes are now returned as Hashes instead of objects with methods.

Changes:
- Add compatibility layer using respond_to? checks
- Support both object-style (.method) and hash-style ([:key]) access
- Update jetstream_bridge.rb fetch_stream_info method
- Update topology/stream.rb ensure_update method
- Update topology/overlap_guard.rb list_streams_with_subjects method
- Update core/debug_helper.rb stream_debug method

This maintains backward compatibility with older nats-pure versions
while supporting the new Hash-based response format in 2.5.0.

Fixes: "undefined method 'streams' for Hash" error during connection

Author: attaradev

CHANGELOG.md

@@ -5,6 +5,17 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [4.0.4] - 2025-11-23
+
+### Fixed
+
+- **NATS Compatibility** - Fix connection failure with nats-pure 2.5.0
+  - Handle both object-style and hash-style access for stream_info responses
+  - Fixes "undefined method 'streams' for Hash" error during connection establishment
+  - Adds compatibility checks using `respond_to?` for config and state attributes
+  - Updated 4 files: jetstream_bridge.rb, topology/stream.rb, topology/overlap_guard.rb, debug_helper.rb
+  - Maintains backward compatibility with older nats-pure versions
+
 ## [4.0.3] - 2025-11-23
 
 ### Added

Gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    jetstream_bridge (4.0.3)
+    jetstream_bridge (4.0.4)
       activerecord (>= 7.1.5.2, < 8.0)
       activesupport (>= 7.1.5.2, < 8.0)
       mutex_m

lib/jetstream_bridge.rb

@@ -115,21 +115,39 @@ def ensure_topology!
       Connection.jetstream
     end
 
-    # Health check for monitoring and readiness probes
+    # Active health check for monitoring and readiness probes
+    #
+    # Performs actual operations to verify system health:
+    # - Checks NATS connection (active: calls account_info API)
+    # - Verifies stream exists and is accessible (active: queries stream info)
+    # - Tests NATS round-trip communication (active: RTT measurement)
     #
     # @return [Hash] Health status including NATS connection, stream, and version
     def health_check
+      start_time = Time.now
       conn_instance = Connection.instance
+
+      # Active check: calls @jts.account_info internally
       connected = conn_instance.connected?
       connected_at = conn_instance.connected_at
 
+      # Active check: queries actual stream from NATS server
       stream_info = fetch_stream_info if connected
 
+      # Active check: measure NATS round-trip time
+      rtt_ms = measure_nats_rtt if connected
+
+      health_check_duration_ms = ((Time.now - start_time) * 1000).round(2)
+
       {
         healthy: connected && stream_info&.fetch(:exists, false),
         nats_connected: connected,
         connected_at: connected_at&.iso8601,
         stream: stream_info,
+        performance: {
+          nats_rtt_ms: rtt_ms,
+          health_check_duration_ms: health_check_duration_ms
+        },
         config: {
           env: config.env,
           app_name: config.app_name,
@@ -281,11 +299,18 @@ def subscribe(handler = nil, run: false, durable_name: nil, batch_size: nil, &bl
     def fetch_stream_info
       jts = Connection.jetstream
       info = jts.stream_info(config.stream_name)
+
+      # Handle both object-style and hash-style access for compatibility
+      config_data = info.config
+      state_data = info.state
+      subjects = config_data.respond_to?(:subjects) ? config_data.subjects : config_data[:subjects]
+      messages = state_data.respond_to?(:messages) ? state_data.messages : state_data[:messages]
+
       {
         exists: true,
         name: config.stream_name,
-        subjects: info.config.subjects,
-        messages: info.state.messages
+        subjects: subjects,
+        messages: messages
       }
     rescue StandardError => e
       {
@@ -295,6 +320,20 @@ def fetch_stream_info
       }
     end
 
+    def measure_nats_rtt
+      # Measure round-trip time using NATS RTT method
+      nc = Connection.nc
+      start = Time.now
+      nc.rtt
+      ((Time.now - start) * 1000).round(2)
+    rescue StandardError => e
+      Logging.warn(
+        "Failed to measure NATS RTT: #{e.class} #{e.message}",
+        tag: 'JetstreamBridge'
+      )
+      nil
+    end
+
     def assign!(cfg, key, val)
       setter = :"#{key}="
       raise ArgumentError, "Unknown configuration option: #{key}" unless cfg.respond_to?(setter)

lib/jetstream_bridge/core/debug_helper.rb

@@ -81,17 +81,29 @@ def stream_debug
         cfg = JetstreamBridge.config
         info = jts.stream_info(cfg.stream_name)
 
+        # Handle both object-style and hash-style access for compatibility
+        config_data = info.config
+        state_data = info.state
+        subjects = config_data.respond_to?(:subjects) ? config_data.subjects : config_data[:subjects]
+        retention = config_data.respond_to?(:retention) ? config_data.retention : config_data[:retention]
+        storage = config_data.respond_to?(:storage) ? config_data.storage : config_data[:storage]
+        max_consumers = config_data.respond_to?(:max_consumers) ? config_data.max_consumers : config_data[:max_consumers]
+        messages = state_data.respond_to?(:messages) ? state_data.messages : state_data[:messages]
+        bytes = state_data.respond_to?(:bytes) ? state_data.bytes : state_data[:bytes]
+        first_seq = state_data.respond_to?(:first_seq) ? state_data.first_seq : state_data[:first_seq]
+        last_seq = state_data.respond_to?(:last_seq) ? state_data.last_seq : state_data[:last_seq]
+
         {
           name: cfg.stream_name,
           exists: true,
-          subjects: info.config.subjects,
-          retention: info.config.retention,
-          storage: info.config.storage,
-          max_consumers: info.config.max_consumers,
-          messages: info.state.messages,
-          bytes: info.state.bytes,
-          first_seq: info.state.first_seq,
-          last_seq: info.state.last_seq
+          subjects: subjects,
+          retention: retention,
+          storage: storage,
+          max_consumers: max_consumers,
+          messages: messages,
+          bytes: bytes,
+          first_seq: first_seq,
+          last_seq: last_seq
         }
       rescue StandardError => e
         {

lib/jetstream_bridge/topology/overlap_guard.rb

@@ -48,7 +48,10 @@ def allowed_subjects(jts, target_name, desired_subjects)
       def list_streams_with_subjects(jts)
         list_stream_names(jts).map do |name|
           info = jts.stream_info(name)
-          { name: name, subjects: Array(info.config.subjects || []) }
+          # Handle both object-style and hash-style access for compatibility
+          config_data = info.config
+          subjects = config_data.respond_to?(:subjects) ? config_data.subjects : config_data[:subjects]
+          { name: name, subjects: Array(subjects || []) }
         end
       end
 

lib/jetstream_bridge/topology/stream.rb

@@ -127,11 +127,15 @@ def ensure!(jts, name, subjects)
       private
 
       def ensure_update(jts, name, info, desired_subjects)
-        existing = StreamSupport.normalize_subjects(info.config.subjects || [])
+        # Handle both object-style and hash-style access for compatibility
+        config_data = info.config
+        subjects = config_data.respond_to?(:subjects) ? config_data.subjects : config_data[:subjects]
+        existing = StreamSupport.normalize_subjects(subjects || [])
         to_add   = StreamSupport.missing_subjects(existing, desired_subjects)
 
         # Retention is immutable; if different, skip all updates to avoid 10052 error.
-        have_ret = info.config.retention.to_s.downcase
+        retention = config_data.respond_to?(:retention) ? config_data.retention : config_data[:retention]
+        have_ret = retention.to_s.downcase
         if have_ret != RETENTION
           StreamSupport.log_retention_mismatch(name, have: have_ret, want: RETENTION)
           return
@@ -140,7 +144,8 @@ def ensure_update(jts, name, info, desired_subjects)
         add_subjects(jts, name, existing, to_add) if to_add.any?
 
         # Storage can be updated; do it without passing retention.
-        have_storage = info.config.storage.to_s.downcase
+        storage = config_data.respond_to?(:storage) ? config_data.storage : config_data[:storage]
+        have_storage = storage.to_s.downcase
         if have_storage != STORAGE
           apply_update(jts, name, existing, storage: STORAGE)
           StreamSupport.log_config_updated(name, storage: STORAGE)

lib/jetstream_bridge/version.rb

@@ -4,5 +4,5 @@
 #
 # Version constant for the gem.
 module JetstreamBridge
-  VERSION = '4.0.3'
+  VERSION = '4.0.4'
 end

spec/jetstream_bridge_spec.rb

@@ -129,6 +129,8 @@
              connected_at: Time.utc(2024, 1, 1, 12, 0, 0))
     end
 
+    let(:mock_nc) { double('NATS::Client', rtt: 0.005) }
+
     let(:stream_info_data) do
       double('StreamInfo',
              config: double(subjects: ['test.*']),
@@ -138,7 +140,9 @@
     before do
       allow(JetstreamBridge::Connection).to receive(:instance).and_return(conn_instance)
       allow(JetstreamBridge::Connection).to receive(:jetstream).and_return(jts)
+      allow(JetstreamBridge::Connection).to receive(:nc).and_return(mock_nc)
       allow(jts).to receive(:stream_info).and_return(stream_info_data)
+      allow(JetstreamBridge::Logging).to receive(:warn)
     end
 
     it 'returns health status hash' do
@@ -147,6 +151,7 @@
       expect(result).to have_key(:healthy)
       expect(result).to have_key(:nats_connected)
       expect(result).to have_key(:stream)
+      expect(result).to have_key(:performance)
       expect(result).to have_key(:config)
       expect(result).to have_key(:version)
     end
@@ -196,6 +201,11 @@
         expect(jts).not_to receive(:stream_info)
         described_class.health_check
       end
+
+      it 'does not measure RTT' do
+        expect(mock_nc).not_to receive(:rtt)
+        described_class.health_check
+      end
     end
 
     context 'when stream does not exist' do
@@ -215,6 +225,39 @@
       end
     end
 
+    it 'includes performance metrics' do
+      result = described_class.health_check
+      expect(result[:performance]).to be_a(Hash)
+      expect(result[:performance][:nats_rtt_ms]).to be_a(Numeric)
+      expect(result[:performance][:health_check_duration_ms]).to be_a(Numeric)
+      expect(result[:performance][:health_check_duration_ms]).to be > 0
+    end
+
+    it 'measures NATS RTT actively' do
+      expect(mock_nc).to receive(:rtt)
+      described_class.health_check
+    end
+
+    context 'when RTT measurement fails' do
+      before do
+        allow(mock_nc).to receive(:rtt).and_raise(StandardError, 'RTT failed')
+      end
+
+      it 'returns nil for RTT and logs warning' do
+        result = described_class.health_check
+        expect(result[:performance][:nats_rtt_ms]).to be_nil
+        expect(JetstreamBridge::Logging).to have_received(:warn).with(
+          /Failed to measure NATS RTT/,
+          tag: 'JetstreamBridge'
+        )
+      end
+
+      it 'still reports as healthy if connection works' do
+        result = described_class.health_check
+        expect(result[:healthy]).to be true
+      end
+    end
+
     context 'when health check raises error' do
       before do
         allow(JetstreamBridge::Connection).to receive(:instance).and_raise(StandardError, 'Connection error')
@@ -498,6 +541,8 @@
              connected_at: nil)
     end
 
+    let(:mock_nc) { double('NATS::Client', rtt: 0.005) }
+
     let(:stream_info_data) do
       double('StreamInfo',
              config: double(subjects: ['test.*']),
@@ -507,6 +552,7 @@
     before do
       allow(JetstreamBridge::Connection).to receive(:instance).and_return(conn_instance)
       allow(JetstreamBridge::Connection).to receive(:jetstream).and_return(jts)
+      allow(JetstreamBridge::Connection).to receive(:nc).and_return(mock_nc)
       allow(jts).to receive(:stream_info).and_return(stream_info_data)
     end