Module: NATS::JetStream::API
- Defined in:
- lib/nats/io/js.rb
Overview
JetStream::API are the types used to interact with the JetStream API.
Defined Under Namespace
Classes: ConsumerConfig, ConsumerInfo, RawStreamMsg, SequenceInfo, StreamConfig, StreamCreateResponse, StreamInfo, StreamState
Constant Summary collapse
- Error =
When the server responds with an error from the JetStream API.
::NATS::JetStream::Error::APIError
Instance Attribute Summary collapse
- #ack_floor ⇒ SequenceInfo
- #ack_policy ⇒ String
- #ack_wait ⇒ Integer
- #bytes ⇒ Integer
- #cluster ⇒ Hash
-
#config ⇒ ConsumerConfig
Consumer configuration.
- #consumer_count ⇒ Integer
-
#consumer_seq ⇒ Integer
The consumer sequence.
-
#created ⇒ String
Time when the consumer was created.
- #deliver_policy ⇒ String
- #delivered ⇒ SequenceInfo
- #durable_name ⇒ String
- #first_seq ⇒ Integer
- #last_seq ⇒ Integer
- #max_ack_pending ⇒ Integer
- #max_deliver ⇒ Integer
- #max_waiting ⇒ Integer
- #messages ⇒ Integer
-
#name ⇒ String
Name of the consumer.
- #num_ack_pending ⇒ Integer
- #num_pending ⇒ Integer
- #num_redelivered ⇒ Integer
- #num_waiting ⇒ Integer
- #replay_policy ⇒ String
-
#stream_name ⇒ String
Name of the stream to which the consumer belongs.
-
#stream_seq ⇒ Integer
The stream sequence.
Instance Attribute Details
#ack_floor ⇒ SequenceInfo
1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 |
# File 'lib/nats/io/js.rb', line 1100 ConsumerInfo = Struct.new(:type, :stream_name, :name, :created, :config, :delivered, :ack_floor, :num_ack_pending, :num_redelivered, :num_waiting, :num_pending, :cluster, :push_bound, keyword_init: true) do def initialize(opts={}) opts[:created] = Time.parse(opts[:created]) opts[:ack_floor] = SequenceInfo.new(opts[:ack_floor]) opts[:delivered] = SequenceInfo.new(opts[:delivered]) opts[:config][:ack_wait] = opts[:config][:ack_wait] / 1_000_000_000 opts[:config] = ConsumerConfig.new(opts[:config]) opts.delete(:cluster) # Filter unrecognized fields just in case. rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) freeze end end |
#ack_policy ⇒ String
1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 |
# File 'lib/nats/io/js.rb', line 1138 ConsumerConfig = Struct.new(:durable_name, :description, :deliver_subject, :deliver_group, :deliver_policy, :opt_start_seq, :opt_start_time, :ack_policy, :ack_wait, :max_deliver, :filter_subject, :replay_policy, :rate_limit_bps, :sample_freq, :max_waiting, :max_ack_pending, :flow_control, :idle_heartbeat, :headers_only, keyword_init: true) do def initialize(opts={}) # Filter unrecognized fields just in case. rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) end def to_json(*args) config = self.to_h config.delete_if { |_k, v| v.nil? } config.to_json(*args) end end |
#ack_wait ⇒ Integer
1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 |
# File 'lib/nats/io/js.rb', line 1138 ConsumerConfig = Struct.new(:durable_name, :description, :deliver_subject, :deliver_group, :deliver_policy, :opt_start_seq, :opt_start_time, :ack_policy, :ack_wait, :max_deliver, :filter_subject, :replay_policy, :rate_limit_bps, :sample_freq, :max_waiting, :max_ack_pending, :flow_control, :idle_heartbeat, :headers_only, keyword_init: true) do def initialize(opts={}) # Filter unrecognized fields just in case. rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) end def to_json(*args) config = self.to_h config.delete_if { |_k, v| v.nil? } config.to_json(*args) end end |
#bytes ⇒ Integer
1255 1256 1257 1258 1259 1260 1261 1262 1263 |
# File 'lib/nats/io/js.rb', line 1255 StreamState = Struct.new(:messages, :bytes, :first_seq, :first_ts, :last_seq, :last_ts, :consumer_count, keyword_init: true) do def initialize(opts={}) rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) end end |
#cluster ⇒ Hash
1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 |
# File 'lib/nats/io/js.rb', line 1100 ConsumerInfo = Struct.new(:type, :stream_name, :name, :created, :config, :delivered, :ack_floor, :num_ack_pending, :num_redelivered, :num_waiting, :num_pending, :cluster, :push_bound, keyword_init: true) do def initialize(opts={}) opts[:created] = Time.parse(opts[:created]) opts[:ack_floor] = SequenceInfo.new(opts[:ack_floor]) opts[:delivered] = SequenceInfo.new(opts[:delivered]) opts[:config][:ack_wait] = opts[:config][:ack_wait] / 1_000_000_000 opts[:config] = ConsumerConfig.new(opts[:config]) opts.delete(:cluster) # Filter unrecognized fields just in case. rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) freeze end end |
#config ⇒ ConsumerConfig
Returns consumer configuration.
1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 |
# File 'lib/nats/io/js.rb', line 1100 ConsumerInfo = Struct.new(:type, :stream_name, :name, :created, :config, :delivered, :ack_floor, :num_ack_pending, :num_redelivered, :num_waiting, :num_pending, :cluster, :push_bound, keyword_init: true) do def initialize(opts={}) opts[:created] = Time.parse(opts[:created]) opts[:ack_floor] = SequenceInfo.new(opts[:ack_floor]) opts[:delivered] = SequenceInfo.new(opts[:delivered]) opts[:config][:ack_wait] = opts[:config][:ack_wait] / 1_000_000_000 opts[:config] = ConsumerConfig.new(opts[:config]) opts.delete(:cluster) # Filter unrecognized fields just in case. rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) freeze end end |
#consumer_count ⇒ Integer
1255 1256 1257 1258 1259 1260 1261 1262 1263 |
# File 'lib/nats/io/js.rb', line 1255 StreamState = Struct.new(:messages, :bytes, :first_seq, :first_ts, :last_seq, :last_ts, :consumer_count, keyword_init: true) do def initialize(opts={}) rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) end end |
#consumer_seq ⇒ Integer
Returns The consumer sequence.
1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 |
# File 'lib/nats/io/js.rb', line 1065 SequenceInfo = Struct.new(:consumer_seq, :stream_seq, :last_active, keyword_init: true) do def initialize(opts={}) # Filter unrecognized fields and freeze. rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) freeze end end |
#created ⇒ String
Returns time when the consumer was created.
1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 |
# File 'lib/nats/io/js.rb', line 1100 ConsumerInfo = Struct.new(:type, :stream_name, :name, :created, :config, :delivered, :ack_floor, :num_ack_pending, :num_redelivered, :num_waiting, :num_pending, :cluster, :push_bound, keyword_init: true) do def initialize(opts={}) opts[:created] = Time.parse(opts[:created]) opts[:ack_floor] = SequenceInfo.new(opts[:ack_floor]) opts[:delivered] = SequenceInfo.new(opts[:delivered]) opts[:config][:ack_wait] = opts[:config][:ack_wait] / 1_000_000_000 opts[:config] = ConsumerConfig.new(opts[:config]) opts.delete(:cluster) # Filter unrecognized fields just in case. rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) freeze end end |
#deliver_policy ⇒ String
1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 |
# File 'lib/nats/io/js.rb', line 1138 ConsumerConfig = Struct.new(:durable_name, :description, :deliver_subject, :deliver_group, :deliver_policy, :opt_start_seq, :opt_start_time, :ack_policy, :ack_wait, :max_deliver, :filter_subject, :replay_policy, :rate_limit_bps, :sample_freq, :max_waiting, :max_ack_pending, :flow_control, :idle_heartbeat, :headers_only, keyword_init: true) do def initialize(opts={}) # Filter unrecognized fields just in case. rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) end def to_json(*args) config = self.to_h config.delete_if { |_k, v| v.nil? } config.to_json(*args) end end |
#delivered ⇒ SequenceInfo
1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 |
# File 'lib/nats/io/js.rb', line 1100 ConsumerInfo = Struct.new(:type, :stream_name, :name, :created, :config, :delivered, :ack_floor, :num_ack_pending, :num_redelivered, :num_waiting, :num_pending, :cluster, :push_bound, keyword_init: true) do def initialize(opts={}) opts[:created] = Time.parse(opts[:created]) opts[:ack_floor] = SequenceInfo.new(opts[:ack_floor]) opts[:delivered] = SequenceInfo.new(opts[:delivered]) opts[:config][:ack_wait] = opts[:config][:ack_wait] / 1_000_000_000 opts[:config] = ConsumerConfig.new(opts[:config]) opts.delete(:cluster) # Filter unrecognized fields just in case. rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) freeze end end |
#durable_name ⇒ String
1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 |
# File 'lib/nats/io/js.rb', line 1138 ConsumerConfig = Struct.new(:durable_name, :description, :deliver_subject, :deliver_group, :deliver_policy, :opt_start_seq, :opt_start_time, :ack_policy, :ack_wait, :max_deliver, :filter_subject, :replay_policy, :rate_limit_bps, :sample_freq, :max_waiting, :max_ack_pending, :flow_control, :idle_heartbeat, :headers_only, keyword_init: true) do def initialize(opts={}) # Filter unrecognized fields just in case. rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) end def to_json(*args) config = self.to_h config.delete_if { |_k, v| v.nil? } config.to_json(*args) end end |
#first_seq ⇒ Integer
1255 1256 1257 1258 1259 1260 1261 1262 1263 |
# File 'lib/nats/io/js.rb', line 1255 StreamState = Struct.new(:messages, :bytes, :first_seq, :first_ts, :last_seq, :last_ts, :consumer_count, keyword_init: true) do def initialize(opts={}) rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) end end |
#last_seq ⇒ Integer
1255 1256 1257 1258 1259 1260 1261 1262 1263 |
# File 'lib/nats/io/js.rb', line 1255 StreamState = Struct.new(:messages, :bytes, :first_seq, :first_ts, :last_seq, :last_ts, :consumer_count, keyword_init: true) do def initialize(opts={}) rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) end end |
#max_ack_pending ⇒ Integer
1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 |
# File 'lib/nats/io/js.rb', line 1138 ConsumerConfig = Struct.new(:durable_name, :description, :deliver_subject, :deliver_group, :deliver_policy, :opt_start_seq, :opt_start_time, :ack_policy, :ack_wait, :max_deliver, :filter_subject, :replay_policy, :rate_limit_bps, :sample_freq, :max_waiting, :max_ack_pending, :flow_control, :idle_heartbeat, :headers_only, keyword_init: true) do def initialize(opts={}) # Filter unrecognized fields just in case. rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) end def to_json(*args) config = self.to_h config.delete_if { |_k, v| v.nil? } config.to_json(*args) end end |
#max_deliver ⇒ Integer
1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 |
# File 'lib/nats/io/js.rb', line 1138 ConsumerConfig = Struct.new(:durable_name, :description, :deliver_subject, :deliver_group, :deliver_policy, :opt_start_seq, :opt_start_time, :ack_policy, :ack_wait, :max_deliver, :filter_subject, :replay_policy, :rate_limit_bps, :sample_freq, :max_waiting, :max_ack_pending, :flow_control, :idle_heartbeat, :headers_only, keyword_init: true) do def initialize(opts={}) # Filter unrecognized fields just in case. rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) end def to_json(*args) config = self.to_h config.delete_if { |_k, v| v.nil? } config.to_json(*args) end end |
#max_waiting ⇒ Integer
1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 |
# File 'lib/nats/io/js.rb', line 1138 ConsumerConfig = Struct.new(:durable_name, :description, :deliver_subject, :deliver_group, :deliver_policy, :opt_start_seq, :opt_start_time, :ack_policy, :ack_wait, :max_deliver, :filter_subject, :replay_policy, :rate_limit_bps, :sample_freq, :max_waiting, :max_ack_pending, :flow_control, :idle_heartbeat, :headers_only, keyword_init: true) do def initialize(opts={}) # Filter unrecognized fields just in case. rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) end def to_json(*args) config = self.to_h config.delete_if { |_k, v| v.nil? } config.to_json(*args) end end |
#messages ⇒ Integer
1255 1256 1257 1258 1259 1260 1261 1262 1263 |
# File 'lib/nats/io/js.rb', line 1255 StreamState = Struct.new(:messages, :bytes, :first_seq, :first_ts, :last_seq, :last_ts, :consumer_count, keyword_init: true) do def initialize(opts={}) rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) end end |
#name ⇒ String
Returns name of the consumer.
1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 |
# File 'lib/nats/io/js.rb', line 1100 ConsumerInfo = Struct.new(:type, :stream_name, :name, :created, :config, :delivered, :ack_floor, :num_ack_pending, :num_redelivered, :num_waiting, :num_pending, :cluster, :push_bound, keyword_init: true) do def initialize(opts={}) opts[:created] = Time.parse(opts[:created]) opts[:ack_floor] = SequenceInfo.new(opts[:ack_floor]) opts[:delivered] = SequenceInfo.new(opts[:delivered]) opts[:config][:ack_wait] = opts[:config][:ack_wait] / 1_000_000_000 opts[:config] = ConsumerConfig.new(opts[:config]) opts.delete(:cluster) # Filter unrecognized fields just in case. rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) freeze end end |
#num_ack_pending ⇒ Integer
1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 |
# File 'lib/nats/io/js.rb', line 1100 ConsumerInfo = Struct.new(:type, :stream_name, :name, :created, :config, :delivered, :ack_floor, :num_ack_pending, :num_redelivered, :num_waiting, :num_pending, :cluster, :push_bound, keyword_init: true) do def initialize(opts={}) opts[:created] = Time.parse(opts[:created]) opts[:ack_floor] = SequenceInfo.new(opts[:ack_floor]) opts[:delivered] = SequenceInfo.new(opts[:delivered]) opts[:config][:ack_wait] = opts[:config][:ack_wait] / 1_000_000_000 opts[:config] = ConsumerConfig.new(opts[:config]) opts.delete(:cluster) # Filter unrecognized fields just in case. rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) freeze end end |
#num_pending ⇒ Integer
1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 |
# File 'lib/nats/io/js.rb', line 1100 ConsumerInfo = Struct.new(:type, :stream_name, :name, :created, :config, :delivered, :ack_floor, :num_ack_pending, :num_redelivered, :num_waiting, :num_pending, :cluster, :push_bound, keyword_init: true) do def initialize(opts={}) opts[:created] = Time.parse(opts[:created]) opts[:ack_floor] = SequenceInfo.new(opts[:ack_floor]) opts[:delivered] = SequenceInfo.new(opts[:delivered]) opts[:config][:ack_wait] = opts[:config][:ack_wait] / 1_000_000_000 opts[:config] = ConsumerConfig.new(opts[:config]) opts.delete(:cluster) # Filter unrecognized fields just in case. rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) freeze end end |
#num_redelivered ⇒ Integer
1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 |
# File 'lib/nats/io/js.rb', line 1100 ConsumerInfo = Struct.new(:type, :stream_name, :name, :created, :config, :delivered, :ack_floor, :num_ack_pending, :num_redelivered, :num_waiting, :num_pending, :cluster, :push_bound, keyword_init: true) do def initialize(opts={}) opts[:created] = Time.parse(opts[:created]) opts[:ack_floor] = SequenceInfo.new(opts[:ack_floor]) opts[:delivered] = SequenceInfo.new(opts[:delivered]) opts[:config][:ack_wait] = opts[:config][:ack_wait] / 1_000_000_000 opts[:config] = ConsumerConfig.new(opts[:config]) opts.delete(:cluster) # Filter unrecognized fields just in case. rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) freeze end end |
#num_waiting ⇒ Integer
1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 |
# File 'lib/nats/io/js.rb', line 1100 ConsumerInfo = Struct.new(:type, :stream_name, :name, :created, :config, :delivered, :ack_floor, :num_ack_pending, :num_redelivered, :num_waiting, :num_pending, :cluster, :push_bound, keyword_init: true) do def initialize(opts={}) opts[:created] = Time.parse(opts[:created]) opts[:ack_floor] = SequenceInfo.new(opts[:ack_floor]) opts[:delivered] = SequenceInfo.new(opts[:delivered]) opts[:config][:ack_wait] = opts[:config][:ack_wait] / 1_000_000_000 opts[:config] = ConsumerConfig.new(opts[:config]) opts.delete(:cluster) # Filter unrecognized fields just in case. rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) freeze end end |
#replay_policy ⇒ String
1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 |
# File 'lib/nats/io/js.rb', line 1138 ConsumerConfig = Struct.new(:durable_name, :description, :deliver_subject, :deliver_group, :deliver_policy, :opt_start_seq, :opt_start_time, :ack_policy, :ack_wait, :max_deliver, :filter_subject, :replay_policy, :rate_limit_bps, :sample_freq, :max_waiting, :max_ack_pending, :flow_control, :idle_heartbeat, :headers_only, keyword_init: true) do def initialize(opts={}) # Filter unrecognized fields just in case. rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) end def to_json(*args) config = self.to_h config.delete_if { |_k, v| v.nil? } config.to_json(*args) end end |
#stream_name ⇒ String
Returns name of the stream to which the consumer belongs.
1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 |
# File 'lib/nats/io/js.rb', line 1100 ConsumerInfo = Struct.new(:type, :stream_name, :name, :created, :config, :delivered, :ack_floor, :num_ack_pending, :num_redelivered, :num_waiting, :num_pending, :cluster, :push_bound, keyword_init: true) do def initialize(opts={}) opts[:created] = Time.parse(opts[:created]) opts[:ack_floor] = SequenceInfo.new(opts[:ack_floor]) opts[:delivered] = SequenceInfo.new(opts[:delivered]) opts[:config][:ack_wait] = opts[:config][:ack_wait] / 1_000_000_000 opts[:config] = ConsumerConfig.new(opts[:config]) opts.delete(:cluster) # Filter unrecognized fields just in case. rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) freeze end end |
#stream_seq ⇒ Integer
Returns The stream sequence.
1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 |
# File 'lib/nats/io/js.rb', line 1065 SequenceInfo = Struct.new(:consumer_seq, :stream_seq, :last_active, keyword_init: true) do def initialize(opts={}) # Filter unrecognized fields and freeze. rem = opts.keys - members opts.delete_if { |k| rem.include?(k) } super(opts) freeze end end |