jsm.go 42 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599
  1. // Copyright 2021-2022 The NATS Authors
  2. // Licensed under the Apache License, Version 2.0 (the "License");
  3. // you may not use this file except in compliance with the License.
  4. // You may obtain a copy of the License at
  5. //
  6. // http://www.apache.org/licenses/LICENSE-2.0
  7. //
  8. // Unless required by applicable law or agreed to in writing, software
  9. // distributed under the License is distributed on an "AS IS" BASIS,
  10. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  11. // See the License for the specific language governing permissions and
  12. // limitations under the License.
  13. package nats
  14. import (
  15. "context"
  16. "encoding/json"
  17. "errors"
  18. "fmt"
  19. "strconv"
  20. "strings"
  21. "time"
  22. )
  23. // JetStreamManager manages JetStream Streams and Consumers.
  24. type JetStreamManager interface {
  25. // AddStream creates a stream.
  26. AddStream(cfg *StreamConfig, opts ...JSOpt) (*StreamInfo, error)
  27. // UpdateStream updates a stream.
  28. UpdateStream(cfg *StreamConfig, opts ...JSOpt) (*StreamInfo, error)
  29. // DeleteStream deletes a stream.
  30. DeleteStream(name string, opts ...JSOpt) error
  31. // StreamInfo retrieves information from a stream.
  32. StreamInfo(stream string, opts ...JSOpt) (*StreamInfo, error)
  33. // PurgeStream purges a stream messages.
  34. PurgeStream(name string, opts ...JSOpt) error
  35. // StreamsInfo can be used to retrieve a list of StreamInfo objects.
  36. // DEPRECATED: Use Streams() instead.
  37. StreamsInfo(opts ...JSOpt) <-chan *StreamInfo
  38. // Streams can be used to retrieve a list of StreamInfo objects.
  39. Streams(opts ...JSOpt) <-chan *StreamInfo
  40. // StreamNames is used to retrieve a list of Stream names.
  41. StreamNames(opts ...JSOpt) <-chan string
  42. // GetMsg retrieves a raw stream message stored in JetStream by sequence number.
  43. // Use options nats.DirectGet() or nats.DirectGetNext() to trigger retrieval
  44. // directly from a distributed group of servers (leader and replicas).
  45. // The stream must have been created/updated with the AllowDirect boolean.
  46. GetMsg(name string, seq uint64, opts ...JSOpt) (*RawStreamMsg, error)
  47. // GetLastMsg retrieves the last raw stream message stored in JetStream by subject.
  48. // Use option nats.DirectGet() to trigger retrieval
  49. // directly from a distributed group of servers (leader and replicas).
  50. // The stream must have been created/updated with the AllowDirect boolean.
  51. GetLastMsg(name, subject string, opts ...JSOpt) (*RawStreamMsg, error)
  52. // DeleteMsg deletes a message from a stream. The message is marked as erased, but its value is not overwritten.
  53. DeleteMsg(name string, seq uint64, opts ...JSOpt) error
  54. // SecureDeleteMsg deletes a message from a stream. The deleted message is overwritten with random data
  55. // As a result, this operation is slower than DeleteMsg()
  56. SecureDeleteMsg(name string, seq uint64, opts ...JSOpt) error
  57. // AddConsumer adds a consumer to a stream.
  58. AddConsumer(stream string, cfg *ConsumerConfig, opts ...JSOpt) (*ConsumerInfo, error)
  59. // UpdateConsumer updates an existing consumer.
  60. UpdateConsumer(stream string, cfg *ConsumerConfig, opts ...JSOpt) (*ConsumerInfo, error)
  61. // DeleteConsumer deletes a consumer.
  62. DeleteConsumer(stream, consumer string, opts ...JSOpt) error
  63. // ConsumerInfo retrieves information of a consumer from a stream.
  64. ConsumerInfo(stream, name string, opts ...JSOpt) (*ConsumerInfo, error)
  65. // ConsumersInfo is used to retrieve a list of ConsumerInfo objects.
  66. // DEPRECATED: Use Consumers() instead.
  67. ConsumersInfo(stream string, opts ...JSOpt) <-chan *ConsumerInfo
  68. // Consumers is used to retrieve a list of ConsumerInfo objects.
  69. Consumers(stream string, opts ...JSOpt) <-chan *ConsumerInfo
  70. // ConsumerNames is used to retrieve a list of Consumer names.
  71. ConsumerNames(stream string, opts ...JSOpt) <-chan string
  72. // AccountInfo retrieves info about the JetStream usage from an account.
  73. AccountInfo(opts ...JSOpt) (*AccountInfo, error)
  74. // StreamNameBySubject returns a stream matching given subject.
  75. StreamNameBySubject(string, ...JSOpt) (string, error)
  76. }
  77. // StreamConfig will determine the properties for a stream.
  78. // There are sensible defaults for most. If no subjects are
  79. // given the name will be used as the only subject.
  80. type StreamConfig struct {
  81. Name string `json:"name"`
  82. Description string `json:"description,omitempty"`
  83. Subjects []string `json:"subjects,omitempty"`
  84. Retention RetentionPolicy `json:"retention"`
  85. MaxConsumers int `json:"max_consumers"`
  86. MaxMsgs int64 `json:"max_msgs"`
  87. MaxBytes int64 `json:"max_bytes"`
  88. Discard DiscardPolicy `json:"discard"`
  89. DiscardNewPerSubject bool `json:"discard_new_per_subject,omitempty"`
  90. MaxAge time.Duration `json:"max_age"`
  91. MaxMsgsPerSubject int64 `json:"max_msgs_per_subject"`
  92. MaxMsgSize int32 `json:"max_msg_size,omitempty"`
  93. Storage StorageType `json:"storage"`
  94. Replicas int `json:"num_replicas"`
  95. NoAck bool `json:"no_ack,omitempty"`
  96. Template string `json:"template_owner,omitempty"`
  97. Duplicates time.Duration `json:"duplicate_window,omitempty"`
  98. Placement *Placement `json:"placement,omitempty"`
  99. Mirror *StreamSource `json:"mirror,omitempty"`
  100. Sources []*StreamSource `json:"sources,omitempty"`
  101. Sealed bool `json:"sealed,omitempty"`
  102. DenyDelete bool `json:"deny_delete,omitempty"`
  103. DenyPurge bool `json:"deny_purge,omitempty"`
  104. AllowRollup bool `json:"allow_rollup_hdrs,omitempty"`
  105. // Allow republish of the message after being sequenced and stored.
  106. RePublish *RePublish `json:"republish,omitempty"`
  107. // Allow higher performance, direct access to get individual messages. E.g. KeyValue
  108. AllowDirect bool `json:"allow_direct"`
  109. // Allow higher performance and unified direct access for mirrors as well.
  110. MirrorDirect bool `json:"mirror_direct"`
  111. }
  112. // RePublish is for republishing messages once committed to a stream. The original
  113. // subject cis remapped from the subject pattern to the destination pattern.
  114. type RePublish struct {
  115. Source string `json:"src,omitempty"`
  116. Destination string `json:"dest"`
  117. HeadersOnly bool `json:"headers_only,omitempty"`
  118. }
  119. // Placement is used to guide placement of streams in clustered JetStream.
  120. type Placement struct {
  121. Cluster string `json:"cluster"`
  122. Tags []string `json:"tags,omitempty"`
  123. }
  124. // StreamSource dictates how streams can source from other streams.
  125. type StreamSource struct {
  126. Name string `json:"name"`
  127. OptStartSeq uint64 `json:"opt_start_seq,omitempty"`
  128. OptStartTime *time.Time `json:"opt_start_time,omitempty"`
  129. FilterSubject string `json:"filter_subject,omitempty"`
  130. External *ExternalStream `json:"external,omitempty"`
  131. Domain string `json:"-"`
  132. }
  133. // ExternalStream allows you to qualify access to a stream source in another
  134. // account.
  135. type ExternalStream struct {
  136. APIPrefix string `json:"api"`
  137. DeliverPrefix string `json:"deliver,omitempty"`
  138. }
  139. // Helper for copying when we do not want to change user's version.
  140. func (ss *StreamSource) copy() *StreamSource {
  141. nss := *ss
  142. // Check pointers
  143. if ss.OptStartTime != nil {
  144. t := *ss.OptStartTime
  145. nss.OptStartTime = &t
  146. }
  147. if ss.External != nil {
  148. ext := *ss.External
  149. nss.External = &ext
  150. }
  151. return &nss
  152. }
  153. // If we have a Domain, convert to the appropriate ext.APIPrefix.
  154. // This will change the stream source, so should be a copy passed in.
  155. func (ss *StreamSource) convertDomain() error {
  156. if ss.Domain == _EMPTY_ {
  157. return nil
  158. }
  159. if ss.External != nil {
  160. // These should be mutually exclusive.
  161. // TODO(dlc) - Make generic?
  162. return errors.New("nats: domain and external are both set")
  163. }
  164. ss.External = &ExternalStream{APIPrefix: fmt.Sprintf(jsExtDomainT, ss.Domain)}
  165. return nil
  166. }
  167. // apiResponse is a standard response from the JetStream JSON API
  168. type apiResponse struct {
  169. Type string `json:"type"`
  170. Error *APIError `json:"error,omitempty"`
  171. }
  172. // apiPaged includes variables used to create paged responses from the JSON API
  173. type apiPaged struct {
  174. Total int `json:"total"`
  175. Offset int `json:"offset"`
  176. Limit int `json:"limit"`
  177. }
  178. // apiPagedRequest includes parameters allowing specific pages to be requested
  179. // from APIs responding with apiPaged.
  180. type apiPagedRequest struct {
  181. Offset int `json:"offset,omitempty"`
  182. }
  183. // AccountInfo contains info about the JetStream usage from the current account.
  184. type AccountInfo struct {
  185. Tier
  186. Domain string `json:"domain"`
  187. API APIStats `json:"api"`
  188. Tiers map[string]Tier `json:"tiers"`
  189. }
  190. type Tier struct {
  191. Memory uint64 `json:"memory"`
  192. Store uint64 `json:"storage"`
  193. Streams int `json:"streams"`
  194. Consumers int `json:"consumers"`
  195. Limits AccountLimits `json:"limits"`
  196. }
  197. // APIStats reports on API calls to JetStream for this account.
  198. type APIStats struct {
  199. Total uint64 `json:"total"`
  200. Errors uint64 `json:"errors"`
  201. }
  202. // AccountLimits includes the JetStream limits of the current account.
  203. type AccountLimits struct {
  204. MaxMemory int64 `json:"max_memory"`
  205. MaxStore int64 `json:"max_storage"`
  206. MaxStreams int `json:"max_streams"`
  207. MaxConsumers int `json:"max_consumers"`
  208. MaxAckPending int `json:"max_ack_pending"`
  209. MemoryMaxStreamBytes int64 `json:"memory_max_stream_bytes"`
  210. StoreMaxStreamBytes int64 `json:"storage_max_stream_bytes"`
  211. MaxBytesRequired bool `json:"max_bytes_required"`
  212. }
  213. type accountInfoResponse struct {
  214. apiResponse
  215. AccountInfo
  216. }
  217. // AccountInfo retrieves info about the JetStream usage from the current account.
  218. // If JetStream is not enabled, this will return ErrJetStreamNotEnabled
  219. // Other errors can happen but are generally considered retryable
  220. func (js *js) AccountInfo(opts ...JSOpt) (*AccountInfo, error) {
  221. o, cancel, err := getJSContextOpts(js.opts, opts...)
  222. if err != nil {
  223. return nil, err
  224. }
  225. if cancel != nil {
  226. defer cancel()
  227. }
  228. resp, err := js.apiRequestWithContext(o.ctx, js.apiSubj(apiAccountInfo), nil)
  229. if err != nil {
  230. // todo maybe nats server should never have no responder on this subject and always respond if they know there is no js to be had
  231. if err == ErrNoResponders {
  232. err = ErrJetStreamNotEnabled
  233. }
  234. return nil, err
  235. }
  236. var info accountInfoResponse
  237. if err := json.Unmarshal(resp.Data, &info); err != nil {
  238. return nil, err
  239. }
  240. if info.Error != nil {
  241. // Internally checks based on error code instead of description match.
  242. if errors.Is(info.Error, ErrJetStreamNotEnabledForAccount) {
  243. return nil, ErrJetStreamNotEnabledForAccount
  244. }
  245. return nil, info.Error
  246. }
  247. return &info.AccountInfo, nil
  248. }
  249. type createConsumerRequest struct {
  250. Stream string `json:"stream_name"`
  251. Config *ConsumerConfig `json:"config"`
  252. }
  253. type consumerResponse struct {
  254. apiResponse
  255. *ConsumerInfo
  256. }
  257. // AddConsumer will add a JetStream consumer.
  258. func (js *js) AddConsumer(stream string, cfg *ConsumerConfig, opts ...JSOpt) (*ConsumerInfo, error) {
  259. if cfg == nil {
  260. cfg = &ConsumerConfig{}
  261. }
  262. consumerName := cfg.Name
  263. if consumerName == _EMPTY_ {
  264. consumerName = cfg.Durable
  265. }
  266. if consumerName != _EMPTY_ {
  267. consInfo, err := js.ConsumerInfo(stream, consumerName, opts...)
  268. if err != nil && !errors.Is(err, ErrConsumerNotFound) && !errors.Is(err, ErrStreamNotFound) {
  269. return nil, err
  270. }
  271. if consInfo != nil {
  272. sameConfig := checkConfig(&consInfo.Config, cfg)
  273. if sameConfig != nil {
  274. return nil, fmt.Errorf("%w: creating consumer %q on stream %q", ErrConsumerNameAlreadyInUse, consumerName, stream)
  275. } else {
  276. return consInfo, nil
  277. }
  278. }
  279. }
  280. return js.upsertConsumer(stream, consumerName, cfg, opts...)
  281. }
  282. func (js *js) UpdateConsumer(stream string, cfg *ConsumerConfig, opts ...JSOpt) (*ConsumerInfo, error) {
  283. if cfg == nil {
  284. return nil, ErrConsumerConfigRequired
  285. }
  286. consumerName := cfg.Name
  287. if consumerName == _EMPTY_ {
  288. consumerName = cfg.Durable
  289. }
  290. if consumerName == _EMPTY_ {
  291. return nil, ErrConsumerNameRequired
  292. }
  293. return js.upsertConsumer(stream, consumerName, cfg, opts...)
  294. }
  295. func (js *js) upsertConsumer(stream, consumerName string, cfg *ConsumerConfig, opts ...JSOpt) (*ConsumerInfo, error) {
  296. if err := checkStreamName(stream); err != nil {
  297. return nil, err
  298. }
  299. o, cancel, err := getJSContextOpts(js.opts, opts...)
  300. if err != nil {
  301. return nil, err
  302. }
  303. if cancel != nil {
  304. defer cancel()
  305. }
  306. req, err := json.Marshal(&createConsumerRequest{Stream: stream, Config: cfg})
  307. if err != nil {
  308. return nil, err
  309. }
  310. var ccSubj string
  311. if consumerName == _EMPTY_ {
  312. // if consumer name is empty (neither Durable nor Name is set), use the legacy ephemeral endpoint
  313. ccSubj = fmt.Sprintf(apiLegacyConsumerCreateT, stream)
  314. } else if err := checkConsumerName(consumerName); err != nil {
  315. return nil, err
  316. } else if js.nc.serverMinVersion(2, 9, 0) {
  317. if cfg.Durable != "" && js.opts.featureFlags.useDurableConsumerCreate {
  318. // if user set the useDurableConsumerCreate flag, use the legacy DURABLE.CREATE endpoint
  319. ccSubj = fmt.Sprintf(apiDurableCreateT, stream, consumerName)
  320. } else if cfg.FilterSubject == _EMPTY_ || cfg.FilterSubject == ">" {
  321. // if filter subject is empty or ">", use the endpoint without filter subject
  322. ccSubj = fmt.Sprintf(apiConsumerCreateT, stream, consumerName)
  323. } else {
  324. // if filter subject is not empty, use the endpoint with filter subject
  325. ccSubj = fmt.Sprintf(apiConsumerCreateWithFilterSubjectT, stream, consumerName, cfg.FilterSubject)
  326. }
  327. } else {
  328. if cfg.Durable != "" {
  329. // if Durable is set, use the DURABLE.CREATE endpoint
  330. ccSubj = fmt.Sprintf(apiDurableCreateT, stream, consumerName)
  331. } else {
  332. // if Durable is not set, use the legacy ephemeral endpoint
  333. ccSubj = fmt.Sprintf(apiLegacyConsumerCreateT, stream)
  334. }
  335. }
  336. resp, err := js.apiRequestWithContext(o.ctx, js.apiSubj(ccSubj), req)
  337. if err != nil {
  338. if err == ErrNoResponders {
  339. err = ErrJetStreamNotEnabled
  340. }
  341. return nil, err
  342. }
  343. var info consumerResponse
  344. err = json.Unmarshal(resp.Data, &info)
  345. if err != nil {
  346. return nil, err
  347. }
  348. if info.Error != nil {
  349. if errors.Is(info.Error, ErrStreamNotFound) {
  350. return nil, ErrStreamNotFound
  351. }
  352. if errors.Is(info.Error, ErrConsumerNotFound) {
  353. return nil, ErrConsumerNotFound
  354. }
  355. return nil, info.Error
  356. }
  357. return info.ConsumerInfo, nil
  358. }
  359. // consumerDeleteResponse is the response for a Consumer delete request.
  360. type consumerDeleteResponse struct {
  361. apiResponse
  362. Success bool `json:"success,omitempty"`
  363. }
  364. func checkStreamName(stream string) error {
  365. if stream == _EMPTY_ {
  366. return ErrStreamNameRequired
  367. }
  368. if strings.ContainsAny(stream, ". ") {
  369. return ErrInvalidStreamName
  370. }
  371. return nil
  372. }
  373. // Check that the consumer name is not empty and is valid (does not contain "." and " ").
  374. // Additional consumer name validation is done in nats-server.
  375. // Returns ErrConsumerNameRequired if consumer name is empty, ErrInvalidConsumerName is invalid, otherwise nil
  376. func checkConsumerName(consumer string) error {
  377. if consumer == _EMPTY_ {
  378. return ErrConsumerNameRequired
  379. }
  380. if strings.ContainsAny(consumer, ". ") {
  381. return ErrInvalidConsumerName
  382. }
  383. return nil
  384. }
  385. // DeleteConsumer deletes a Consumer.
  386. func (js *js) DeleteConsumer(stream, consumer string, opts ...JSOpt) error {
  387. if err := checkStreamName(stream); err != nil {
  388. return err
  389. }
  390. if err := checkConsumerName(consumer); err != nil {
  391. return err
  392. }
  393. o, cancel, err := getJSContextOpts(js.opts, opts...)
  394. if err != nil {
  395. return err
  396. }
  397. if cancel != nil {
  398. defer cancel()
  399. }
  400. dcSubj := js.apiSubj(fmt.Sprintf(apiConsumerDeleteT, stream, consumer))
  401. r, err := js.apiRequestWithContext(o.ctx, dcSubj, nil)
  402. if err != nil {
  403. return err
  404. }
  405. var resp consumerDeleteResponse
  406. if err := json.Unmarshal(r.Data, &resp); err != nil {
  407. return err
  408. }
  409. if resp.Error != nil {
  410. if errors.Is(resp.Error, ErrConsumerNotFound) {
  411. return ErrConsumerNotFound
  412. }
  413. return resp.Error
  414. }
  415. return nil
  416. }
  417. // ConsumerInfo returns information about a Consumer.
  418. func (js *js) ConsumerInfo(stream, consumer string, opts ...JSOpt) (*ConsumerInfo, error) {
  419. if err := checkStreamName(stream); err != nil {
  420. return nil, err
  421. }
  422. if err := checkConsumerName(consumer); err != nil {
  423. return nil, err
  424. }
  425. o, cancel, err := getJSContextOpts(js.opts, opts...)
  426. if err != nil {
  427. return nil, err
  428. }
  429. if cancel != nil {
  430. defer cancel()
  431. }
  432. return js.getConsumerInfoContext(o.ctx, stream, consumer)
  433. }
  434. // consumerLister fetches pages of ConsumerInfo objects. This object is not
  435. // safe to use for multiple threads.
  436. type consumerLister struct {
  437. stream string
  438. js *js
  439. err error
  440. offset int
  441. page []*ConsumerInfo
  442. pageInfo *apiPaged
  443. }
  444. // consumersRequest is the type used for Consumers requests.
  445. type consumersRequest struct {
  446. apiPagedRequest
  447. }
  448. // consumerListResponse is the response for a Consumers List request.
  449. type consumerListResponse struct {
  450. apiResponse
  451. apiPaged
  452. Consumers []*ConsumerInfo `json:"consumers"`
  453. }
  454. // Next fetches the next ConsumerInfo page.
  455. func (c *consumerLister) Next() bool {
  456. if c.err != nil {
  457. return false
  458. }
  459. if err := checkStreamName(c.stream); err != nil {
  460. c.err = err
  461. return false
  462. }
  463. if c.pageInfo != nil && c.offset >= c.pageInfo.Total {
  464. return false
  465. }
  466. req, err := json.Marshal(consumersRequest{
  467. apiPagedRequest: apiPagedRequest{Offset: c.offset},
  468. })
  469. if err != nil {
  470. c.err = err
  471. return false
  472. }
  473. var cancel context.CancelFunc
  474. ctx := c.js.opts.ctx
  475. if ctx == nil {
  476. ctx, cancel = context.WithTimeout(context.Background(), c.js.opts.wait)
  477. defer cancel()
  478. }
  479. clSubj := c.js.apiSubj(fmt.Sprintf(apiConsumerListT, c.stream))
  480. r, err := c.js.apiRequestWithContext(ctx, clSubj, req)
  481. if err != nil {
  482. c.err = err
  483. return false
  484. }
  485. var resp consumerListResponse
  486. if err := json.Unmarshal(r.Data, &resp); err != nil {
  487. c.err = err
  488. return false
  489. }
  490. if resp.Error != nil {
  491. c.err = resp.Error
  492. return false
  493. }
  494. c.pageInfo = &resp.apiPaged
  495. c.page = resp.Consumers
  496. c.offset += len(c.page)
  497. return true
  498. }
  499. // Page returns the current ConsumerInfo page.
  500. func (c *consumerLister) Page() []*ConsumerInfo {
  501. return c.page
  502. }
  503. // Err returns any errors found while fetching pages.
  504. func (c *consumerLister) Err() error {
  505. return c.err
  506. }
  507. // Consumers is used to retrieve a list of ConsumerInfo objects.
  508. func (jsc *js) Consumers(stream string, opts ...JSOpt) <-chan *ConsumerInfo {
  509. o, cancel, err := getJSContextOpts(jsc.opts, opts...)
  510. if err != nil {
  511. return nil
  512. }
  513. ch := make(chan *ConsumerInfo)
  514. l := &consumerLister{js: &js{nc: jsc.nc, opts: o}, stream: stream}
  515. go func() {
  516. if cancel != nil {
  517. defer cancel()
  518. }
  519. defer close(ch)
  520. for l.Next() {
  521. for _, info := range l.Page() {
  522. select {
  523. case ch <- info:
  524. case <-o.ctx.Done():
  525. return
  526. }
  527. }
  528. }
  529. }()
  530. return ch
  531. }
  532. // ConsumersInfo is used to retrieve a list of ConsumerInfo objects.
  533. // DEPRECATED: Use Consumers() instead.
  534. func (jsc *js) ConsumersInfo(stream string, opts ...JSOpt) <-chan *ConsumerInfo {
  535. return jsc.Consumers(stream, opts...)
  536. }
  537. type consumerNamesLister struct {
  538. stream string
  539. js *js
  540. err error
  541. offset int
  542. page []string
  543. pageInfo *apiPaged
  544. }
  545. // consumerNamesListResponse is the response for a Consumers Names List request.
  546. type consumerNamesListResponse struct {
  547. apiResponse
  548. apiPaged
  549. Consumers []string `json:"consumers"`
  550. }
  551. // Next fetches the next consumer names page.
  552. func (c *consumerNamesLister) Next() bool {
  553. if c.err != nil {
  554. return false
  555. }
  556. if err := checkStreamName(c.stream); err != nil {
  557. c.err = err
  558. return false
  559. }
  560. if c.pageInfo != nil && c.offset >= c.pageInfo.Total {
  561. return false
  562. }
  563. var cancel context.CancelFunc
  564. ctx := c.js.opts.ctx
  565. if ctx == nil {
  566. ctx, cancel = context.WithTimeout(context.Background(), c.js.opts.wait)
  567. defer cancel()
  568. }
  569. req, err := json.Marshal(consumersRequest{
  570. apiPagedRequest: apiPagedRequest{Offset: c.offset},
  571. })
  572. if err != nil {
  573. c.err = err
  574. return false
  575. }
  576. clSubj := c.js.apiSubj(fmt.Sprintf(apiConsumerNamesT, c.stream))
  577. r, err := c.js.apiRequestWithContext(ctx, clSubj, req)
  578. if err != nil {
  579. c.err = err
  580. return false
  581. }
  582. var resp consumerNamesListResponse
  583. if err := json.Unmarshal(r.Data, &resp); err != nil {
  584. c.err = err
  585. return false
  586. }
  587. if resp.Error != nil {
  588. c.err = resp.Error
  589. return false
  590. }
  591. c.pageInfo = &resp.apiPaged
  592. c.page = resp.Consumers
  593. c.offset += len(c.page)
  594. return true
  595. }
  596. // Page returns the current ConsumerInfo page.
  597. func (c *consumerNamesLister) Page() []string {
  598. return c.page
  599. }
  600. // Err returns any errors found while fetching pages.
  601. func (c *consumerNamesLister) Err() error {
  602. return c.err
  603. }
  604. // ConsumerNames is used to retrieve a list of Consumer names.
  605. func (jsc *js) ConsumerNames(stream string, opts ...JSOpt) <-chan string {
  606. o, cancel, err := getJSContextOpts(jsc.opts, opts...)
  607. if err != nil {
  608. return nil
  609. }
  610. ch := make(chan string)
  611. l := &consumerNamesLister{stream: stream, js: &js{nc: jsc.nc, opts: o}}
  612. go func() {
  613. if cancel != nil {
  614. defer cancel()
  615. }
  616. defer close(ch)
  617. for l.Next() {
  618. for _, info := range l.Page() {
  619. select {
  620. case ch <- info:
  621. case <-o.ctx.Done():
  622. return
  623. }
  624. }
  625. }
  626. }()
  627. return ch
  628. }
  629. // streamCreateResponse stream creation.
  630. type streamCreateResponse struct {
  631. apiResponse
  632. *StreamInfo
  633. }
  634. func (js *js) AddStream(cfg *StreamConfig, opts ...JSOpt) (*StreamInfo, error) {
  635. if cfg == nil {
  636. return nil, ErrStreamConfigRequired
  637. }
  638. if err := checkStreamName(cfg.Name); err != nil {
  639. return nil, err
  640. }
  641. o, cancel, err := getJSContextOpts(js.opts, opts...)
  642. if err != nil {
  643. return nil, err
  644. }
  645. if cancel != nil {
  646. defer cancel()
  647. }
  648. // In case we need to change anything, copy so we do not change the caller's version.
  649. ncfg := *cfg
  650. // If we have a mirror and an external domain, convert to ext.APIPrefix.
  651. if cfg.Mirror != nil && cfg.Mirror.Domain != _EMPTY_ {
  652. // Copy so we do not change the caller's version.
  653. ncfg.Mirror = ncfg.Mirror.copy()
  654. if err := ncfg.Mirror.convertDomain(); err != nil {
  655. return nil, err
  656. }
  657. }
  658. // Check sources for the same.
  659. if len(ncfg.Sources) > 0 {
  660. ncfg.Sources = append([]*StreamSource(nil), ncfg.Sources...)
  661. for i, ss := range ncfg.Sources {
  662. if ss.Domain != _EMPTY_ {
  663. ncfg.Sources[i] = ss.copy()
  664. if err := ncfg.Sources[i].convertDomain(); err != nil {
  665. return nil, err
  666. }
  667. }
  668. }
  669. }
  670. req, err := json.Marshal(&ncfg)
  671. if err != nil {
  672. return nil, err
  673. }
  674. csSubj := js.apiSubj(fmt.Sprintf(apiStreamCreateT, cfg.Name))
  675. r, err := js.apiRequestWithContext(o.ctx, csSubj, req)
  676. if err != nil {
  677. return nil, err
  678. }
  679. var resp streamCreateResponse
  680. if err := json.Unmarshal(r.Data, &resp); err != nil {
  681. return nil, err
  682. }
  683. if resp.Error != nil {
  684. if errors.Is(resp.Error, ErrStreamNameAlreadyInUse) {
  685. return nil, ErrStreamNameAlreadyInUse
  686. }
  687. return nil, resp.Error
  688. }
  689. return resp.StreamInfo, nil
  690. }
  691. type (
  692. // StreamInfoRequest contains additional option to return
  693. StreamInfoRequest struct {
  694. apiPagedRequest
  695. // DeletedDetails when true includes information about deleted messages
  696. DeletedDetails bool `json:"deleted_details,omitempty"`
  697. // SubjectsFilter when set, returns information on the matched subjects
  698. SubjectsFilter string `json:"subjects_filter,omitempty"`
  699. }
  700. streamInfoResponse = struct {
  701. apiResponse
  702. apiPaged
  703. *StreamInfo
  704. }
  705. )
  706. func (js *js) StreamInfo(stream string, opts ...JSOpt) (*StreamInfo, error) {
  707. if err := checkStreamName(stream); err != nil {
  708. return nil, err
  709. }
  710. o, cancel, err := getJSContextOpts(js.opts, opts...)
  711. if err != nil {
  712. return nil, err
  713. }
  714. if cancel != nil {
  715. defer cancel()
  716. }
  717. var i int
  718. var subjectMessagesMap map[string]uint64
  719. var req []byte
  720. var requestPayload bool
  721. var siOpts StreamInfoRequest
  722. if o.streamInfoOpts != nil {
  723. requestPayload = true
  724. siOpts = *o.streamInfoOpts
  725. }
  726. for {
  727. if requestPayload {
  728. siOpts.Offset = i
  729. if req, err = json.Marshal(&siOpts); err != nil {
  730. return nil, err
  731. }
  732. }
  733. siSubj := js.apiSubj(fmt.Sprintf(apiStreamInfoT, stream))
  734. r, err := js.apiRequestWithContext(o.ctx, siSubj, req)
  735. if err != nil {
  736. return nil, err
  737. }
  738. var resp streamInfoResponse
  739. if err := json.Unmarshal(r.Data, &resp); err != nil {
  740. return nil, err
  741. }
  742. if resp.Error != nil {
  743. if errors.Is(resp.Error, ErrStreamNotFound) {
  744. return nil, ErrStreamNotFound
  745. }
  746. return nil, resp.Error
  747. }
  748. var total int
  749. // for backwards compatibility
  750. if resp.Total != 0 {
  751. total = resp.Total
  752. } else {
  753. total = len(resp.State.Subjects)
  754. }
  755. if requestPayload && len(resp.StreamInfo.State.Subjects) > 0 {
  756. if subjectMessagesMap == nil {
  757. subjectMessagesMap = make(map[string]uint64, total)
  758. }
  759. for k, j := range resp.State.Subjects {
  760. subjectMessagesMap[k] = j
  761. i++
  762. }
  763. }
  764. if i >= total {
  765. if requestPayload {
  766. resp.StreamInfo.State.Subjects = subjectMessagesMap
  767. }
  768. return resp.StreamInfo, nil
  769. }
  770. }
  771. }
  772. // StreamInfo shows config and current state for this stream.
  773. type StreamInfo struct {
  774. Config StreamConfig `json:"config"`
  775. Created time.Time `json:"created"`
  776. State StreamState `json:"state"`
  777. Cluster *ClusterInfo `json:"cluster,omitempty"`
  778. Mirror *StreamSourceInfo `json:"mirror,omitempty"`
  779. Sources []*StreamSourceInfo `json:"sources,omitempty"`
  780. Alternates []*StreamAlternate `json:"alternates,omitempty"`
  781. }
  782. // StreamAlternate is an alternate stream represented by a mirror.
  783. type StreamAlternate struct {
  784. Name string `json:"name"`
  785. Domain string `json:"domain,omitempty"`
  786. Cluster string `json:"cluster"`
  787. }
  788. // StreamSourceInfo shows information about an upstream stream source.
  789. type StreamSourceInfo struct {
  790. Name string `json:"name"`
  791. Lag uint64 `json:"lag"`
  792. Active time.Duration `json:"active"`
  793. External *ExternalStream `json:"external"`
  794. Error *APIError `json:"error"`
  795. }
  796. // StreamState is information about the given stream.
  797. type StreamState struct {
  798. Msgs uint64 `json:"messages"`
  799. Bytes uint64 `json:"bytes"`
  800. FirstSeq uint64 `json:"first_seq"`
  801. FirstTime time.Time `json:"first_ts"`
  802. LastSeq uint64 `json:"last_seq"`
  803. LastTime time.Time `json:"last_ts"`
  804. Consumers int `json:"consumer_count"`
  805. Deleted []uint64 `json:"deleted"`
  806. NumDeleted int `json:"num_deleted"`
  807. NumSubjects uint64 `json:"num_subjects"`
  808. Subjects map[string]uint64 `json:"subjects"`
  809. }
  810. // ClusterInfo shows information about the underlying set of servers
  811. // that make up the stream or consumer.
  812. type ClusterInfo struct {
  813. Name string `json:"name,omitempty"`
  814. Leader string `json:"leader,omitempty"`
  815. Replicas []*PeerInfo `json:"replicas,omitempty"`
  816. }
  817. // PeerInfo shows information about all the peers in the cluster that
  818. // are supporting the stream or consumer.
  819. type PeerInfo struct {
  820. Name string `json:"name"`
  821. Current bool `json:"current"`
  822. Offline bool `json:"offline,omitempty"`
  823. Active time.Duration `json:"active"`
  824. Lag uint64 `json:"lag,omitempty"`
  825. }
  826. // UpdateStream updates a Stream.
  827. func (js *js) UpdateStream(cfg *StreamConfig, opts ...JSOpt) (*StreamInfo, error) {
  828. if cfg == nil {
  829. return nil, ErrStreamConfigRequired
  830. }
  831. if err := checkStreamName(cfg.Name); err != nil {
  832. return nil, err
  833. }
  834. o, cancel, err := getJSContextOpts(js.opts, opts...)
  835. if err != nil {
  836. return nil, err
  837. }
  838. if cancel != nil {
  839. defer cancel()
  840. }
  841. req, err := json.Marshal(cfg)
  842. if err != nil {
  843. return nil, err
  844. }
  845. usSubj := js.apiSubj(fmt.Sprintf(apiStreamUpdateT, cfg.Name))
  846. r, err := js.apiRequestWithContext(o.ctx, usSubj, req)
  847. if err != nil {
  848. return nil, err
  849. }
  850. var resp streamInfoResponse
  851. if err := json.Unmarshal(r.Data, &resp); err != nil {
  852. return nil, err
  853. }
  854. if resp.Error != nil {
  855. if errors.Is(resp.Error, ErrStreamNotFound) {
  856. return nil, ErrStreamNotFound
  857. }
  858. return nil, resp.Error
  859. }
  860. return resp.StreamInfo, nil
  861. }
  862. // streamDeleteResponse is the response for a Stream delete request.
  863. type streamDeleteResponse struct {
  864. apiResponse
  865. Success bool `json:"success,omitempty"`
  866. }
  867. // DeleteStream deletes a Stream.
  868. func (js *js) DeleteStream(name string, opts ...JSOpt) error {
  869. if err := checkStreamName(name); err != nil {
  870. return err
  871. }
  872. o, cancel, err := getJSContextOpts(js.opts, opts...)
  873. if err != nil {
  874. return err
  875. }
  876. if cancel != nil {
  877. defer cancel()
  878. }
  879. dsSubj := js.apiSubj(fmt.Sprintf(apiStreamDeleteT, name))
  880. r, err := js.apiRequestWithContext(o.ctx, dsSubj, nil)
  881. if err != nil {
  882. return err
  883. }
  884. var resp streamDeleteResponse
  885. if err := json.Unmarshal(r.Data, &resp); err != nil {
  886. return err
  887. }
  888. if resp.Error != nil {
  889. if errors.Is(resp.Error, ErrStreamNotFound) {
  890. return ErrStreamNotFound
  891. }
  892. return resp.Error
  893. }
  894. return nil
  895. }
  896. type apiMsgGetRequest struct {
  897. Seq uint64 `json:"seq,omitempty"`
  898. LastFor string `json:"last_by_subj,omitempty"`
  899. NextFor string `json:"next_by_subj,omitempty"`
  900. }
  901. // RawStreamMsg is a raw message stored in JetStream.
  902. type RawStreamMsg struct {
  903. Subject string
  904. Sequence uint64
  905. Header Header
  906. Data []byte
  907. Time time.Time
  908. }
  909. // storedMsg is a raw message stored in JetStream.
  910. type storedMsg struct {
  911. Subject string `json:"subject"`
  912. Sequence uint64 `json:"seq"`
  913. Header []byte `json:"hdrs,omitempty"`
  914. Data []byte `json:"data,omitempty"`
  915. Time time.Time `json:"time"`
  916. }
  917. // apiMsgGetResponse is the response for a Stream get request.
  918. type apiMsgGetResponse struct {
  919. apiResponse
  920. Message *storedMsg `json:"message,omitempty"`
  921. }
  922. // GetLastMsg retrieves the last raw stream message stored in JetStream by subject.
  923. func (js *js) GetLastMsg(name, subject string, opts ...JSOpt) (*RawStreamMsg, error) {
  924. return js.getMsg(name, &apiMsgGetRequest{LastFor: subject}, opts...)
  925. }
  926. // GetMsg retrieves a raw stream message stored in JetStream by sequence number.
  927. func (js *js) GetMsg(name string, seq uint64, opts ...JSOpt) (*RawStreamMsg, error) {
  928. return js.getMsg(name, &apiMsgGetRequest{Seq: seq}, opts...)
  929. }
  930. // Low level getMsg
  931. func (js *js) getMsg(name string, mreq *apiMsgGetRequest, opts ...JSOpt) (*RawStreamMsg, error) {
  932. o, cancel, err := getJSContextOpts(js.opts, opts...)
  933. if err != nil {
  934. return nil, err
  935. }
  936. if cancel != nil {
  937. defer cancel()
  938. }
  939. if err := checkStreamName(name); err != nil {
  940. return nil, err
  941. }
  942. var apiSubj string
  943. if o.directGet && mreq.LastFor != _EMPTY_ {
  944. apiSubj = apiDirectMsgGetLastBySubjectT
  945. dsSubj := js.apiSubj(fmt.Sprintf(apiSubj, name, mreq.LastFor))
  946. r, err := js.apiRequestWithContext(o.ctx, dsSubj, nil)
  947. if err != nil {
  948. return nil, err
  949. }
  950. return convertDirectGetMsgResponseToMsg(name, r)
  951. }
  952. if o.directGet {
  953. apiSubj = apiDirectMsgGetT
  954. mreq.NextFor = o.directNextFor
  955. } else {
  956. apiSubj = apiMsgGetT
  957. }
  958. req, err := json.Marshal(mreq)
  959. if err != nil {
  960. return nil, err
  961. }
  962. dsSubj := js.apiSubj(fmt.Sprintf(apiSubj, name))
  963. r, err := js.apiRequestWithContext(o.ctx, dsSubj, req)
  964. if err != nil {
  965. return nil, err
  966. }
  967. if o.directGet {
  968. return convertDirectGetMsgResponseToMsg(name, r)
  969. }
  970. var resp apiMsgGetResponse
  971. if err := json.Unmarshal(r.Data, &resp); err != nil {
  972. return nil, err
  973. }
  974. if resp.Error != nil {
  975. if errors.Is(resp.Error, ErrMsgNotFound) {
  976. return nil, ErrMsgNotFound
  977. }
  978. if errors.Is(resp.Error, ErrStreamNotFound) {
  979. return nil, ErrStreamNotFound
  980. }
  981. return nil, resp.Error
  982. }
  983. msg := resp.Message
  984. var hdr Header
  985. if len(msg.Header) > 0 {
  986. hdr, err = DecodeHeadersMsg(msg.Header)
  987. if err != nil {
  988. return nil, err
  989. }
  990. }
  991. return &RawStreamMsg{
  992. Subject: msg.Subject,
  993. Sequence: msg.Sequence,
  994. Header: hdr,
  995. Data: msg.Data,
  996. Time: msg.Time,
  997. }, nil
  998. }
  999. func convertDirectGetMsgResponseToMsg(name string, r *Msg) (*RawStreamMsg, error) {
  1000. // Check for 404/408. We would get a no-payload message and a "Status" header
  1001. if len(r.Data) == 0 {
  1002. val := r.Header.Get(statusHdr)
  1003. if val != _EMPTY_ {
  1004. switch val {
  1005. case noMessagesSts:
  1006. return nil, ErrMsgNotFound
  1007. default:
  1008. desc := r.Header.Get(descrHdr)
  1009. if desc == _EMPTY_ {
  1010. desc = "unable to get message"
  1011. }
  1012. return nil, fmt.Errorf("nats: %s", desc)
  1013. }
  1014. }
  1015. }
  1016. // Check for headers that give us the required information to
  1017. // reconstruct the message.
  1018. if len(r.Header) == 0 {
  1019. return nil, fmt.Errorf("nats: response should have headers")
  1020. }
  1021. stream := r.Header.Get(JSStream)
  1022. if stream == _EMPTY_ {
  1023. return nil, fmt.Errorf("nats: missing stream header")
  1024. }
  1025. // Mirrors can now answer direct gets, so removing check for name equality.
  1026. // TODO(dlc) - We could have server also have a header with origin and check that?
  1027. seqStr := r.Header.Get(JSSequence)
  1028. if seqStr == _EMPTY_ {
  1029. return nil, fmt.Errorf("nats: missing sequence header")
  1030. }
  1031. seq, err := strconv.ParseUint(seqStr, 10, 64)
  1032. if err != nil {
  1033. return nil, fmt.Errorf("nats: invalid sequence header '%s': %v", seqStr, err)
  1034. }
  1035. timeStr := r.Header.Get(JSTimeStamp)
  1036. if timeStr == _EMPTY_ {
  1037. return nil, fmt.Errorf("nats: missing timestamp header")
  1038. }
  1039. // Temporary code: the server in main branch is sending with format
  1040. // "2006-01-02 15:04:05.999999999 +0000 UTC", but will be changed
  1041. // to use format RFC3339Nano. Because of server test deps/cycle,
  1042. // support both until the server PR lands.
  1043. tm, err := time.Parse(time.RFC3339Nano, timeStr)
  1044. if err != nil {
  1045. tm, err = time.Parse("2006-01-02 15:04:05.999999999 +0000 UTC", timeStr)
  1046. if err != nil {
  1047. return nil, fmt.Errorf("nats: invalid timestamp header '%s': %v", timeStr, err)
  1048. }
  1049. }
  1050. subj := r.Header.Get(JSSubject)
  1051. if subj == _EMPTY_ {
  1052. return nil, fmt.Errorf("nats: missing subject header")
  1053. }
  1054. return &RawStreamMsg{
  1055. Subject: subj,
  1056. Sequence: seq,
  1057. Header: r.Header,
  1058. Data: r.Data,
  1059. Time: tm,
  1060. }, nil
  1061. }
  1062. type msgDeleteRequest struct {
  1063. Seq uint64 `json:"seq"`
  1064. NoErase bool `json:"no_erase,omitempty"`
  1065. }
  1066. // msgDeleteResponse is the response for a Stream delete request.
  1067. type msgDeleteResponse struct {
  1068. apiResponse
  1069. Success bool `json:"success,omitempty"`
  1070. }
  1071. // DeleteMsg deletes a message from a stream.
  1072. // The message is marked as erased, but not overwritten
  1073. func (js *js) DeleteMsg(name string, seq uint64, opts ...JSOpt) error {
  1074. o, cancel, err := getJSContextOpts(js.opts, opts...)
  1075. if err != nil {
  1076. return err
  1077. }
  1078. if cancel != nil {
  1079. defer cancel()
  1080. }
  1081. return js.deleteMsg(o.ctx, name, &msgDeleteRequest{Seq: seq, NoErase: true})
  1082. }
  1083. // SecureDeleteMsg deletes a message from a stream. The deleted message is overwritten with random data
  1084. // As a result, this operation is slower than DeleteMsg()
  1085. func (js *js) SecureDeleteMsg(name string, seq uint64, opts ...JSOpt) error {
  1086. o, cancel, err := getJSContextOpts(js.opts, opts...)
  1087. if err != nil {
  1088. return err
  1089. }
  1090. if cancel != nil {
  1091. defer cancel()
  1092. }
  1093. return js.deleteMsg(o.ctx, name, &msgDeleteRequest{Seq: seq})
  1094. }
  1095. func (js *js) deleteMsg(ctx context.Context, stream string, req *msgDeleteRequest) error {
  1096. if err := checkStreamName(stream); err != nil {
  1097. return err
  1098. }
  1099. reqJSON, err := json.Marshal(req)
  1100. if err != nil {
  1101. return err
  1102. }
  1103. dsSubj := js.apiSubj(fmt.Sprintf(apiMsgDeleteT, stream))
  1104. r, err := js.apiRequestWithContext(ctx, dsSubj, reqJSON)
  1105. if err != nil {
  1106. return err
  1107. }
  1108. var resp msgDeleteResponse
  1109. if err := json.Unmarshal(r.Data, &resp); err != nil {
  1110. return err
  1111. }
  1112. if resp.Error != nil {
  1113. return resp.Error
  1114. }
  1115. return nil
  1116. }
  1117. // StreamPurgeRequest is optional request information to the purge API.
  1118. type StreamPurgeRequest struct {
  1119. // Purge up to but not including sequence.
  1120. Sequence uint64 `json:"seq,omitempty"`
  1121. // Subject to match against messages for the purge command.
  1122. Subject string `json:"filter,omitempty"`
  1123. // Number of messages to keep.
  1124. Keep uint64 `json:"keep,omitempty"`
  1125. }
  1126. type streamPurgeResponse struct {
  1127. apiResponse
  1128. Success bool `json:"success,omitempty"`
  1129. Purged uint64 `json:"purged"`
  1130. }
  1131. // PurgeStream purges messages on a Stream.
  1132. func (js *js) PurgeStream(stream string, opts ...JSOpt) error {
  1133. if err := checkStreamName(stream); err != nil {
  1134. return err
  1135. }
  1136. var req *StreamPurgeRequest
  1137. var ok bool
  1138. for _, opt := range opts {
  1139. // For PurgeStream, only request body opt is relevant
  1140. if req, ok = opt.(*StreamPurgeRequest); ok {
  1141. break
  1142. }
  1143. }
  1144. return js.purgeStream(stream, req)
  1145. }
  1146. func (js *js) purgeStream(stream string, req *StreamPurgeRequest, opts ...JSOpt) error {
  1147. o, cancel, err := getJSContextOpts(js.opts, opts...)
  1148. if err != nil {
  1149. return err
  1150. }
  1151. if cancel != nil {
  1152. defer cancel()
  1153. }
  1154. var b []byte
  1155. if req != nil {
  1156. if b, err = json.Marshal(req); err != nil {
  1157. return err
  1158. }
  1159. }
  1160. psSubj := js.apiSubj(fmt.Sprintf(apiStreamPurgeT, stream))
  1161. r, err := js.apiRequestWithContext(o.ctx, psSubj, b)
  1162. if err != nil {
  1163. return err
  1164. }
  1165. var resp streamPurgeResponse
  1166. if err := json.Unmarshal(r.Data, &resp); err != nil {
  1167. return err
  1168. }
  1169. if resp.Error != nil {
  1170. if errors.Is(resp.Error, ErrBadRequest) {
  1171. return fmt.Errorf("%w: %s", ErrBadRequest, "invalid purge request body")
  1172. }
  1173. return resp.Error
  1174. }
  1175. return nil
  1176. }
  1177. // streamLister fetches pages of StreamInfo objects. This object is not safe
  1178. // to use for multiple threads.
  1179. type streamLister struct {
  1180. js *js
  1181. page []*StreamInfo
  1182. err error
  1183. offset int
  1184. pageInfo *apiPaged
  1185. }
  1186. // streamListResponse list of detailed stream information.
  1187. // A nil request is valid and means all streams.
  1188. type streamListResponse struct {
  1189. apiResponse
  1190. apiPaged
  1191. Streams []*StreamInfo `json:"streams"`
  1192. }
  1193. // streamNamesRequest is used for Stream Name requests.
  1194. type streamNamesRequest struct {
  1195. apiPagedRequest
  1196. // These are filters that can be applied to the list.
  1197. Subject string `json:"subject,omitempty"`
  1198. }
  1199. // Next fetches the next StreamInfo page.
  1200. func (s *streamLister) Next() bool {
  1201. if s.err != nil {
  1202. return false
  1203. }
  1204. if s.pageInfo != nil && s.offset >= s.pageInfo.Total {
  1205. return false
  1206. }
  1207. req, err := json.Marshal(streamNamesRequest{
  1208. apiPagedRequest: apiPagedRequest{Offset: s.offset},
  1209. Subject: s.js.opts.streamListSubject,
  1210. })
  1211. if err != nil {
  1212. s.err = err
  1213. return false
  1214. }
  1215. var cancel context.CancelFunc
  1216. ctx := s.js.opts.ctx
  1217. if ctx == nil {
  1218. ctx, cancel = context.WithTimeout(context.Background(), s.js.opts.wait)
  1219. defer cancel()
  1220. }
  1221. slSubj := s.js.apiSubj(apiStreamListT)
  1222. r, err := s.js.apiRequestWithContext(ctx, slSubj, req)
  1223. if err != nil {
  1224. s.err = err
  1225. return false
  1226. }
  1227. var resp streamListResponse
  1228. if err := json.Unmarshal(r.Data, &resp); err != nil {
  1229. s.err = err
  1230. return false
  1231. }
  1232. if resp.Error != nil {
  1233. s.err = resp.Error
  1234. return false
  1235. }
  1236. s.pageInfo = &resp.apiPaged
  1237. s.page = resp.Streams
  1238. s.offset += len(s.page)
  1239. return true
  1240. }
  1241. // Page returns the current StreamInfo page.
  1242. func (s *streamLister) Page() []*StreamInfo {
  1243. return s.page
  1244. }
  1245. // Err returns any errors found while fetching pages.
  1246. func (s *streamLister) Err() error {
  1247. return s.err
  1248. }
  1249. // Streams can be used to retrieve a list of StreamInfo objects.
  1250. func (jsc *js) Streams(opts ...JSOpt) <-chan *StreamInfo {
  1251. o, cancel, err := getJSContextOpts(jsc.opts, opts...)
  1252. if err != nil {
  1253. return nil
  1254. }
  1255. ch := make(chan *StreamInfo)
  1256. l := &streamLister{js: &js{nc: jsc.nc, opts: o}}
  1257. go func() {
  1258. if cancel != nil {
  1259. defer cancel()
  1260. }
  1261. defer close(ch)
  1262. for l.Next() {
  1263. for _, info := range l.Page() {
  1264. select {
  1265. case ch <- info:
  1266. case <-o.ctx.Done():
  1267. return
  1268. }
  1269. }
  1270. }
  1271. }()
  1272. return ch
  1273. }
  1274. // StreamsInfo can be used to retrieve a list of StreamInfo objects.
  1275. // DEPRECATED: Use Streams() instead.
  1276. func (jsc *js) StreamsInfo(opts ...JSOpt) <-chan *StreamInfo {
  1277. return jsc.Streams(opts...)
  1278. }
  1279. type streamNamesLister struct {
  1280. js *js
  1281. err error
  1282. offset int
  1283. page []string
  1284. pageInfo *apiPaged
  1285. }
  1286. // Next fetches the next stream names page.
  1287. func (l *streamNamesLister) Next() bool {
  1288. if l.err != nil {
  1289. return false
  1290. }
  1291. if l.pageInfo != nil && l.offset >= l.pageInfo.Total {
  1292. return false
  1293. }
  1294. var cancel context.CancelFunc
  1295. ctx := l.js.opts.ctx
  1296. if ctx == nil {
  1297. ctx, cancel = context.WithTimeout(context.Background(), l.js.opts.wait)
  1298. defer cancel()
  1299. }
  1300. req, err := json.Marshal(streamNamesRequest{
  1301. apiPagedRequest: apiPagedRequest{Offset: l.offset},
  1302. Subject: l.js.opts.streamListSubject,
  1303. })
  1304. if err != nil {
  1305. l.err = err
  1306. return false
  1307. }
  1308. r, err := l.js.apiRequestWithContext(ctx, l.js.apiSubj(apiStreams), req)
  1309. if err != nil {
  1310. l.err = err
  1311. return false
  1312. }
  1313. var resp streamNamesResponse
  1314. if err := json.Unmarshal(r.Data, &resp); err != nil {
  1315. l.err = err
  1316. return false
  1317. }
  1318. if resp.Error != nil {
  1319. l.err = resp.Error
  1320. return false
  1321. }
  1322. l.pageInfo = &resp.apiPaged
  1323. l.page = resp.Streams
  1324. l.offset += len(l.page)
  1325. return true
  1326. }
  1327. // Page returns the current ConsumerInfo page.
  1328. func (l *streamNamesLister) Page() []string {
  1329. return l.page
  1330. }
  1331. // Err returns any errors found while fetching pages.
  1332. func (l *streamNamesLister) Err() error {
  1333. return l.err
  1334. }
  1335. // StreamNames is used to retrieve a list of Stream names.
  1336. func (jsc *js) StreamNames(opts ...JSOpt) <-chan string {
  1337. o, cancel, err := getJSContextOpts(jsc.opts, opts...)
  1338. if err != nil {
  1339. return nil
  1340. }
  1341. ch := make(chan string)
  1342. l := &streamNamesLister{js: &js{nc: jsc.nc, opts: o}}
  1343. go func() {
  1344. if cancel != nil {
  1345. defer cancel()
  1346. }
  1347. defer close(ch)
  1348. for l.Next() {
  1349. for _, info := range l.Page() {
  1350. select {
  1351. case ch <- info:
  1352. case <-o.ctx.Done():
  1353. return
  1354. }
  1355. }
  1356. }
  1357. }()
  1358. return ch
  1359. }
  1360. // StreamNameBySubject returns a stream name that matches the subject.
  1361. func (jsc *js) StreamNameBySubject(subj string, opts ...JSOpt) (string, error) {
  1362. o, cancel, err := getJSContextOpts(jsc.opts, opts...)
  1363. if err != nil {
  1364. return "", err
  1365. }
  1366. if cancel != nil {
  1367. defer cancel()
  1368. }
  1369. var slr streamNamesResponse
  1370. req := &streamRequest{subj}
  1371. j, err := json.Marshal(req)
  1372. if err != nil {
  1373. return _EMPTY_, err
  1374. }
  1375. resp, err := jsc.apiRequestWithContext(o.ctx, jsc.apiSubj(apiStreams), j)
  1376. if err != nil {
  1377. if err == ErrNoResponders {
  1378. err = ErrJetStreamNotEnabled
  1379. }
  1380. return _EMPTY_, err
  1381. }
  1382. if err := json.Unmarshal(resp.Data, &slr); err != nil {
  1383. return _EMPTY_, err
  1384. }
  1385. if slr.Error != nil || len(slr.Streams) != 1 {
  1386. return _EMPTY_, ErrNoMatchingStream
  1387. }
  1388. return slr.Streams[0], nil
  1389. }
  1390. func getJSContextOpts(defs *jsOpts, opts ...JSOpt) (*jsOpts, context.CancelFunc, error) {
  1391. var o jsOpts
  1392. for _, opt := range opts {
  1393. if err := opt.configureJSContext(&o); err != nil {
  1394. return nil, nil, err
  1395. }
  1396. }
  1397. // Check for option collisions. Right now just timeout and context.
  1398. if o.ctx != nil && o.wait != 0 {
  1399. return nil, nil, ErrContextAndTimeout
  1400. }
  1401. if o.wait == 0 && o.ctx == nil {
  1402. o.wait = defs.wait
  1403. }
  1404. var cancel context.CancelFunc
  1405. if o.ctx == nil && o.wait > 0 {
  1406. o.ctx, cancel = context.WithTimeout(context.Background(), o.wait)
  1407. }
  1408. if o.pre == _EMPTY_ {
  1409. o.pre = defs.pre
  1410. }
  1411. return &o, cancel, nil
  1412. }