js.go 104 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427242824292430243124322433243424352436243724382439244024412442244324442445244624472448244924502451245224532454245524562457245824592460246124622463246424652466246724682469247024712472247324742475247624772478247924802481248224832484248524862487248824892490249124922493249424952496249724982499250025012502250325042505250625072508250925102511251225132514251525162517251825192520252125222523252425252526252725282529253025312532253325342535253625372538253925402541254225432544254525462547254825492550255125522553255425552556255725582559256025612562256325642565256625672568256925702571257225732574257525762577257825792580258125822583258425852586258725882589259025912592259325942595259625972598259926002601260226032604260526062607260826092610261126122613261426152616261726182619262026212622262326242625262626272628262926302631263226332634263526362637263826392640264126422643264426452646264726482649265026512652265326542655265626572658265926602661266226632664266526662667266826692670267126722673267426752676267726782679268026812682268326842685268626872688268926902691269226932694269526962697269826992700270127022703270427052706270727082709271027112712271327142715271627172718271927202721272227232724272527262727272827292730273127322733273427352736273727382739274027412742274327442745274627472748274927502751275227532754275527562757275827592760276127622763276427652766276727682769277027712772277327742775277627772778277927802781278227832784278527862787278827892790279127922793279427952796279727982799280028012802280328042805280628072808280928102811281228132814281528162817281828192820282128222823282428252826282728282829283028312832283328342835283628372838283928402841284228432844284528462847284828492850285128522853285428552856285728582859286028612862286328642865286628672868286928702871287228732874287528762877287828792880288128822883288428852886288728882889289028912892289328942895289628972898289929002901290229032904290529062907290829092910291129122913291429152916291729182919292029212922292329242925292629272928292929302931293229332934293529362937293829392940294129422943294429452946294729482949295029512952295329542955295629572958295929602961296229632964296529662967296829692970297129722973297429752976297729782979298029812982298329842985298629872988298929902991299229932994299529962997299829993000300130023003300430053006300730083009301030113012301330143015301630173018301930203021302230233024302530263027302830293030303130323033303430353036303730383039304030413042304330443045304630473048304930503051305230533054305530563057305830593060306130623063306430653066306730683069307030713072307330743075307630773078307930803081308230833084308530863087308830893090309130923093309430953096309730983099310031013102310331043105310631073108310931103111311231133114311531163117311831193120312131223123312431253126312731283129313031313132313331343135313631373138313931403141314231433144314531463147314831493150315131523153315431553156315731583159316031613162316331643165316631673168316931703171317231733174317531763177317831793180318131823183318431853186318731883189319031913192319331943195319631973198319932003201320232033204320532063207320832093210321132123213321432153216321732183219322032213222322332243225322632273228322932303231323232333234323532363237323832393240324132423243324432453246324732483249325032513252325332543255325632573258325932603261326232633264326532663267326832693270327132723273327432753276327732783279328032813282328332843285328632873288328932903291329232933294329532963297329832993300330133023303330433053306330733083309331033113312331333143315331633173318331933203321332233233324332533263327332833293330333133323333333433353336333733383339334033413342334333443345334633473348334933503351335233533354335533563357335833593360336133623363336433653366336733683369337033713372337333743375337633773378337933803381338233833384338533863387338833893390339133923393339433953396339733983399340034013402340334043405340634073408340934103411341234133414341534163417341834193420342134223423342434253426342734283429343034313432343334343435343634373438343934403441344234433444344534463447344834493450345134523453345434553456345734583459346034613462346334643465346634673468346934703471347234733474347534763477347834793480348134823483348434853486348734883489349034913492349334943495349634973498349935003501350235033504350535063507350835093510351135123513351435153516351735183519352035213522352335243525352635273528352935303531353235333534353535363537353835393540354135423543354435453546354735483549355035513552355335543555355635573558355935603561356235633564356535663567356835693570357135723573357435753576357735783579358035813582358335843585358635873588358935903591359235933594359535963597359835993600360136023603360436053606360736083609361036113612361336143615361636173618361936203621362236233624362536263627362836293630363136323633363436353636363736383639364036413642364336443645364636473648364936503651365236533654365536563657365836593660366136623663366436653666
  1. // Copyright 2020-2023 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. "bytes"
  16. "context"
  17. "crypto/sha256"
  18. "encoding/json"
  19. "errors"
  20. "fmt"
  21. "math/rand"
  22. "strconv"
  23. "strings"
  24. "sync"
  25. "sync/atomic"
  26. "time"
  27. "github.com/nats-io/nats.go/internal/parser"
  28. "github.com/nats-io/nuid"
  29. )
  30. // JetStream allows persistent messaging through JetStream.
  31. type JetStream interface {
  32. // Publish publishes a message to JetStream.
  33. Publish(subj string, data []byte, opts ...PubOpt) (*PubAck, error)
  34. // PublishMsg publishes a Msg to JetStream.
  35. PublishMsg(m *Msg, opts ...PubOpt) (*PubAck, error)
  36. // PublishAsync publishes a message to JetStream and returns a PubAckFuture.
  37. // The data should not be changed until the PubAckFuture has been processed.
  38. PublishAsync(subj string, data []byte, opts ...PubOpt) (PubAckFuture, error)
  39. // PublishMsgAsync publishes a Msg to JetStream and returns a PubAckFuture.
  40. // The message should not be changed until the PubAckFuture has been processed.
  41. PublishMsgAsync(m *Msg, opts ...PubOpt) (PubAckFuture, error)
  42. // PublishAsyncPending returns the number of async publishes outstanding for this context.
  43. PublishAsyncPending() int
  44. // PublishAsyncComplete returns a channel that will be closed when all outstanding messages are ack'd.
  45. PublishAsyncComplete() <-chan struct{}
  46. // Subscribe creates an async Subscription for JetStream.
  47. // The stream and consumer names can be provided with the nats.Bind() option.
  48. // For creating an ephemeral (where the consumer name is picked by the server),
  49. // you can provide the stream name with nats.BindStream().
  50. // If no stream name is specified, the library will attempt to figure out which
  51. // stream the subscription is for. See important notes below for more details.
  52. //
  53. // IMPORTANT NOTES:
  54. // * If none of the options Bind() nor Durable() are specified, the library will
  55. // send a request to the server to create an ephemeral JetStream consumer,
  56. // which will be deleted after an Unsubscribe() or Drain(), or automatically
  57. // by the server after a short period of time after the NATS subscription is
  58. // gone.
  59. // * If Durable() option is specified, the library will attempt to lookup a JetStream
  60. // consumer with this name, and if found, will bind to it and not attempt to
  61. // delete it. However, if not found, the library will send a request to
  62. // create such durable JetStream consumer. Note that the library will delete
  63. // the JetStream consumer after an Unsubscribe() or Drain() only if it
  64. // created the durable consumer while subscribing. If the durable consumer
  65. // already existed prior to subscribing it won't be deleted.
  66. // * If Bind() option is provided, the library will attempt to lookup the
  67. // consumer with the given name, and if successful, bind to it. If the lookup fails,
  68. // then the Subscribe() call will return an error.
  69. Subscribe(subj string, cb MsgHandler, opts ...SubOpt) (*Subscription, error)
  70. // SubscribeSync creates a Subscription that can be used to process messages synchronously.
  71. // See important note in Subscribe()
  72. SubscribeSync(subj string, opts ...SubOpt) (*Subscription, error)
  73. // ChanSubscribe creates channel based Subscription.
  74. // See important note in Subscribe()
  75. ChanSubscribe(subj string, ch chan *Msg, opts ...SubOpt) (*Subscription, error)
  76. // ChanQueueSubscribe creates channel based Subscription with a queue group.
  77. // See important note in QueueSubscribe()
  78. ChanQueueSubscribe(subj, queue string, ch chan *Msg, opts ...SubOpt) (*Subscription, error)
  79. // QueueSubscribe creates a Subscription with a queue group.
  80. // If no optional durable name nor binding options are specified, the queue name will be used as a durable name.
  81. // See important note in Subscribe()
  82. QueueSubscribe(subj, queue string, cb MsgHandler, opts ...SubOpt) (*Subscription, error)
  83. // QueueSubscribeSync creates a Subscription with a queue group that can be used to process messages synchronously.
  84. // See important note in QueueSubscribe()
  85. QueueSubscribeSync(subj, queue string, opts ...SubOpt) (*Subscription, error)
  86. // PullSubscribe creates a Subscription that can fetch messages.
  87. // See important note in Subscribe(). Additionally, for an ephemeral pull consumer, the "durable" value must be
  88. // set to an empty string.
  89. PullSubscribe(subj, durable string, opts ...SubOpt) (*Subscription, error)
  90. }
  91. // JetStreamContext allows JetStream messaging and stream management.
  92. type JetStreamContext interface {
  93. JetStream
  94. JetStreamManager
  95. KeyValueManager
  96. ObjectStoreManager
  97. }
  98. // Request API subjects for JetStream.
  99. const (
  100. // defaultAPIPrefix is the default prefix for the JetStream API.
  101. defaultAPIPrefix = "$JS.API."
  102. // jsDomainT is used to create JetStream API prefix by specifying only Domain
  103. jsDomainT = "$JS.%s.API."
  104. // jsExtDomainT is used to create a StreamSource External APIPrefix
  105. jsExtDomainT = "$JS.%s.API"
  106. // apiAccountInfo is for obtaining general information about JetStream.
  107. apiAccountInfo = "INFO"
  108. // apiConsumerCreateT is used to create consumers.
  109. // it accepts stream name and consumer name.
  110. apiConsumerCreateT = "CONSUMER.CREATE.%s.%s"
  111. // apiConsumerCreateT is used to create consumers.
  112. // it accepts stream name, consumer name and filter subject
  113. apiConsumerCreateWithFilterSubjectT = "CONSUMER.CREATE.%s.%s.%s"
  114. // apiLegacyConsumerCreateT is used to create consumers.
  115. // this is a legacy endpoint to support creating ephemerals before nats-server v2.9.0.
  116. apiLegacyConsumerCreateT = "CONSUMER.CREATE.%s"
  117. // apiDurableCreateT is used to create durable consumers.
  118. // this is a legacy endpoint to support creating durable consumers before nats-server v2.9.0.
  119. apiDurableCreateT = "CONSUMER.DURABLE.CREATE.%s.%s"
  120. // apiConsumerInfoT is used to create consumers.
  121. apiConsumerInfoT = "CONSUMER.INFO.%s.%s"
  122. // apiRequestNextT is the prefix for the request next message(s) for a consumer in worker/pull mode.
  123. apiRequestNextT = "CONSUMER.MSG.NEXT.%s.%s"
  124. // apiConsumerDeleteT is used to delete consumers.
  125. apiConsumerDeleteT = "CONSUMER.DELETE.%s.%s"
  126. // apiConsumerListT is used to return all detailed consumer information
  127. apiConsumerListT = "CONSUMER.LIST.%s"
  128. // apiConsumerNamesT is used to return a list with all consumer names for the stream.
  129. apiConsumerNamesT = "CONSUMER.NAMES.%s"
  130. // apiStreams can lookup a stream by subject.
  131. apiStreams = "STREAM.NAMES"
  132. // apiStreamCreateT is the endpoint to create new streams.
  133. apiStreamCreateT = "STREAM.CREATE.%s"
  134. // apiStreamInfoT is the endpoint to get information on a stream.
  135. apiStreamInfoT = "STREAM.INFO.%s"
  136. // apiStreamUpdateT is the endpoint to update existing streams.
  137. apiStreamUpdateT = "STREAM.UPDATE.%s"
  138. // apiStreamDeleteT is the endpoint to delete streams.
  139. apiStreamDeleteT = "STREAM.DELETE.%s"
  140. // apiStreamPurgeT is the endpoint to purge streams.
  141. apiStreamPurgeT = "STREAM.PURGE.%s"
  142. // apiStreamListT is the endpoint that will return all detailed stream information
  143. apiStreamListT = "STREAM.LIST"
  144. // apiMsgGetT is the endpoint to get a message.
  145. apiMsgGetT = "STREAM.MSG.GET.%s"
  146. // apiMsgGetT is the endpoint to perform a direct get of a message.
  147. apiDirectMsgGetT = "DIRECT.GET.%s"
  148. // apiDirectMsgGetLastBySubjectT is the endpoint to perform a direct get of a message by subject.
  149. apiDirectMsgGetLastBySubjectT = "DIRECT.GET.%s.%s"
  150. // apiMsgDeleteT is the endpoint to remove a message.
  151. apiMsgDeleteT = "STREAM.MSG.DELETE.%s"
  152. // orderedHeartbeatsInterval is how fast we want HBs from the server during idle.
  153. orderedHeartbeatsInterval = 5 * time.Second
  154. // Scale for threshold of missed HBs or lack of activity.
  155. hbcThresh = 2
  156. // For ChanSubscription, we can't update sub.delivered as we do for other
  157. // type of subscriptions, since the channel is user provided.
  158. // With flow control in play, we will check for flow control on incoming
  159. // messages (as opposed to when they are delivered), but also from a go
  160. // routine. Without this, the subscription would possibly stall until
  161. // a new message or heartbeat/fc are received.
  162. chanSubFCCheckInterval = 250 * time.Millisecond
  163. // Default time wait between retries on Publish iff err is NoResponders.
  164. DefaultPubRetryWait = 250 * time.Millisecond
  165. // Default number of retries
  166. DefaultPubRetryAttempts = 2
  167. // defaultAsyncPubAckInflight is the number of async pub acks inflight.
  168. defaultAsyncPubAckInflight = 4000
  169. )
  170. // Types of control messages, so far heartbeat and flow control
  171. const (
  172. jsCtrlHB = 1
  173. jsCtrlFC = 2
  174. )
  175. // js is an internal struct from a JetStreamContext.
  176. type js struct {
  177. nc *Conn
  178. opts *jsOpts
  179. // For async publish context.
  180. mu sync.RWMutex
  181. rpre string
  182. rsub *Subscription
  183. pafs map[string]*pubAckFuture
  184. stc chan struct{}
  185. dch chan struct{}
  186. rr *rand.Rand
  187. connStatusCh chan (Status)
  188. }
  189. type jsOpts struct {
  190. ctx context.Context
  191. // For importing JetStream from other accounts.
  192. pre string
  193. // Amount of time to wait for API requests.
  194. wait time.Duration
  195. // For async publish error handling.
  196. aecb MsgErrHandler
  197. // Max async pub ack in flight
  198. maxpa int
  199. // the domain that produced the pre
  200. domain string
  201. // enables protocol tracing
  202. ctrace ClientTrace
  203. shouldTrace bool
  204. // purgeOpts contains optional stream purge options
  205. purgeOpts *StreamPurgeRequest
  206. // streamInfoOpts contains optional stream info options
  207. streamInfoOpts *StreamInfoRequest
  208. // streamListSubject is used for subject filtering when listing streams / stream names
  209. streamListSubject string
  210. // For direct get message requests
  211. directGet bool
  212. // For direct get next message
  213. directNextFor string
  214. // featureFlags are used to enable/disable specific JetStream features
  215. featureFlags featureFlags
  216. }
  217. const (
  218. defaultRequestWait = 5 * time.Second
  219. defaultAccountCheck = 20 * time.Second
  220. )
  221. // JetStream returns a JetStreamContext for messaging and stream management.
  222. // Errors are only returned if inconsistent options are provided.
  223. func (nc *Conn) JetStream(opts ...JSOpt) (JetStreamContext, error) {
  224. js := &js{
  225. nc: nc,
  226. opts: &jsOpts{
  227. pre: defaultAPIPrefix,
  228. wait: defaultRequestWait,
  229. maxpa: defaultAsyncPubAckInflight,
  230. },
  231. }
  232. for _, opt := range opts {
  233. if err := opt.configureJSContext(js.opts); err != nil {
  234. return nil, err
  235. }
  236. }
  237. return js, nil
  238. }
  239. // JSOpt configures a JetStreamContext.
  240. type JSOpt interface {
  241. configureJSContext(opts *jsOpts) error
  242. }
  243. // jsOptFn configures an option for the JetStreamContext.
  244. type jsOptFn func(opts *jsOpts) error
  245. func (opt jsOptFn) configureJSContext(opts *jsOpts) error {
  246. return opt(opts)
  247. }
  248. type featureFlags struct {
  249. useDurableConsumerCreate bool
  250. }
  251. // UseLegacyDurableConsumers makes JetStream use the legacy (pre nats-server v2.9.0) subjects for consumer creation.
  252. // If this option is used when creating JetStremContext, $JS.API.CONSUMER.DURABLE.CREATE.<stream>.<consumer> will be used
  253. // to create a consumer with Durable provided, rather than $JS.API.CONSUMER.CREATE.<stream>.<consumer>.
  254. func UseLegacyDurableConsumers() JSOpt {
  255. return jsOptFn(func(opts *jsOpts) error {
  256. opts.featureFlags.useDurableConsumerCreate = true
  257. return nil
  258. })
  259. }
  260. // ClientTrace can be used to trace API interactions for the JetStream Context.
  261. type ClientTrace struct {
  262. RequestSent func(subj string, payload []byte)
  263. ResponseReceived func(subj string, payload []byte, hdr Header)
  264. }
  265. func (ct ClientTrace) configureJSContext(js *jsOpts) error {
  266. js.ctrace = ct
  267. js.shouldTrace = true
  268. return nil
  269. }
  270. // Domain changes the domain part of JetStream API prefix.
  271. func Domain(domain string) JSOpt {
  272. if domain == _EMPTY_ {
  273. return APIPrefix(_EMPTY_)
  274. }
  275. return jsOptFn(func(js *jsOpts) error {
  276. js.domain = domain
  277. js.pre = fmt.Sprintf(jsDomainT, domain)
  278. return nil
  279. })
  280. }
  281. func (s *StreamPurgeRequest) configureJSContext(js *jsOpts) error {
  282. js.purgeOpts = s
  283. return nil
  284. }
  285. func (s *StreamInfoRequest) configureJSContext(js *jsOpts) error {
  286. js.streamInfoOpts = s
  287. return nil
  288. }
  289. // APIPrefix changes the default prefix used for the JetStream API.
  290. func APIPrefix(pre string) JSOpt {
  291. return jsOptFn(func(js *jsOpts) error {
  292. if pre == _EMPTY_ {
  293. return nil
  294. }
  295. js.pre = pre
  296. if !strings.HasSuffix(js.pre, ".") {
  297. js.pre = js.pre + "."
  298. }
  299. return nil
  300. })
  301. }
  302. // DirectGet is an option that can be used to make GetMsg() or GetLastMsg()
  303. // retrieve message directly from a group of servers (leader and replicas)
  304. // if the stream was created with the AllowDirect option.
  305. func DirectGet() JSOpt {
  306. return jsOptFn(func(js *jsOpts) error {
  307. js.directGet = true
  308. return nil
  309. })
  310. }
  311. // DirectGetNext is an option that can be used to make GetMsg() retrieve message
  312. // directly from a group of servers (leader and replicas) if the stream was
  313. // created with the AllowDirect option.
  314. // The server will find the next message matching the filter `subject` starting
  315. // at the start sequence (argument in GetMsg()). The filter `subject` can be a
  316. // wildcard.
  317. func DirectGetNext(subject string) JSOpt {
  318. return jsOptFn(func(js *jsOpts) error {
  319. js.directGet = true
  320. js.directNextFor = subject
  321. return nil
  322. })
  323. }
  324. // StreamListFilter is an option that can be used to configure `StreamsInfo()` and `StreamNames()` requests.
  325. // It allows filtering the retured streams by subject associated with each stream.
  326. // Wildcards can be used. For example, `StreamListFilter(FOO.*.A) will return
  327. // all streams which have at least one subject matching the provided pattern (e.g. FOO.TEST.A).
  328. func StreamListFilter(subject string) JSOpt {
  329. return jsOptFn(func(opts *jsOpts) error {
  330. opts.streamListSubject = subject
  331. return nil
  332. })
  333. }
  334. func (js *js) apiSubj(subj string) string {
  335. if js.opts.pre == _EMPTY_ {
  336. return subj
  337. }
  338. var b strings.Builder
  339. b.WriteString(js.opts.pre)
  340. b.WriteString(subj)
  341. return b.String()
  342. }
  343. // PubOpt configures options for publishing JetStream messages.
  344. type PubOpt interface {
  345. configurePublish(opts *pubOpts) error
  346. }
  347. // pubOptFn is a function option used to configure JetStream Publish.
  348. type pubOptFn func(opts *pubOpts) error
  349. func (opt pubOptFn) configurePublish(opts *pubOpts) error {
  350. return opt(opts)
  351. }
  352. type pubOpts struct {
  353. ctx context.Context
  354. ttl time.Duration
  355. id string
  356. lid string // Expected last msgId
  357. str string // Expected stream name
  358. seq *uint64 // Expected last sequence
  359. lss *uint64 // Expected last sequence per subject
  360. // Publish retries for NoResponders err.
  361. rwait time.Duration // Retry wait between attempts
  362. rnum int // Retry attempts
  363. // stallWait is the max wait of a async pub ack.
  364. stallWait time.Duration
  365. }
  366. // pubAckResponse is the ack response from the JetStream API when publishing a message.
  367. type pubAckResponse struct {
  368. apiResponse
  369. *PubAck
  370. }
  371. // PubAck is an ack received after successfully publishing a message.
  372. type PubAck struct {
  373. Stream string `json:"stream"`
  374. Sequence uint64 `json:"seq"`
  375. Duplicate bool `json:"duplicate,omitempty"`
  376. Domain string `json:"domain,omitempty"`
  377. }
  378. // Headers for published messages.
  379. const (
  380. MsgIdHdr = "Nats-Msg-Id"
  381. ExpectedStreamHdr = "Nats-Expected-Stream"
  382. ExpectedLastSeqHdr = "Nats-Expected-Last-Sequence"
  383. ExpectedLastSubjSeqHdr = "Nats-Expected-Last-Subject-Sequence"
  384. ExpectedLastMsgIdHdr = "Nats-Expected-Last-Msg-Id"
  385. MsgRollup = "Nats-Rollup"
  386. )
  387. // Headers for republished messages and direct gets.
  388. const (
  389. JSStream = "Nats-Stream"
  390. JSSequence = "Nats-Sequence"
  391. JSTimeStamp = "Nats-Time-Stamp"
  392. JSSubject = "Nats-Subject"
  393. JSLastSequence = "Nats-Last-Sequence"
  394. )
  395. // MsgSize is a header that will be part of a consumer's delivered message if HeadersOnly requested.
  396. const MsgSize = "Nats-Msg-Size"
  397. // Rollups, can be subject only or all messages.
  398. const (
  399. MsgRollupSubject = "sub"
  400. MsgRollupAll = "all"
  401. )
  402. // PublishMsg publishes a Msg to a stream from JetStream.
  403. func (js *js) PublishMsg(m *Msg, opts ...PubOpt) (*PubAck, error) {
  404. var o = pubOpts{rwait: DefaultPubRetryWait, rnum: DefaultPubRetryAttempts}
  405. if len(opts) > 0 {
  406. if m.Header == nil {
  407. m.Header = Header{}
  408. }
  409. for _, opt := range opts {
  410. if err := opt.configurePublish(&o); err != nil {
  411. return nil, err
  412. }
  413. }
  414. }
  415. // Check for option collisions. Right now just timeout and context.
  416. if o.ctx != nil && o.ttl != 0 {
  417. return nil, ErrContextAndTimeout
  418. }
  419. if o.ttl == 0 && o.ctx == nil {
  420. o.ttl = js.opts.wait
  421. }
  422. if o.stallWait > 0 {
  423. return nil, fmt.Errorf("nats: stall wait cannot be set to sync publish")
  424. }
  425. if o.id != _EMPTY_ {
  426. m.Header.Set(MsgIdHdr, o.id)
  427. }
  428. if o.lid != _EMPTY_ {
  429. m.Header.Set(ExpectedLastMsgIdHdr, o.lid)
  430. }
  431. if o.str != _EMPTY_ {
  432. m.Header.Set(ExpectedStreamHdr, o.str)
  433. }
  434. if o.seq != nil {
  435. m.Header.Set(ExpectedLastSeqHdr, strconv.FormatUint(*o.seq, 10))
  436. }
  437. if o.lss != nil {
  438. m.Header.Set(ExpectedLastSubjSeqHdr, strconv.FormatUint(*o.lss, 10))
  439. }
  440. var resp *Msg
  441. var err error
  442. if o.ttl > 0 {
  443. resp, err = js.nc.RequestMsg(m, time.Duration(o.ttl))
  444. } else {
  445. resp, err = js.nc.RequestMsgWithContext(o.ctx, m)
  446. }
  447. if err != nil {
  448. for r, ttl := 0, o.ttl; err == ErrNoResponders && (r < o.rnum || o.rnum < 0); r++ {
  449. // To protect against small blips in leadership changes etc, if we get a no responders here retry.
  450. if o.ctx != nil {
  451. select {
  452. case <-o.ctx.Done():
  453. case <-time.After(o.rwait):
  454. }
  455. } else {
  456. time.Sleep(o.rwait)
  457. }
  458. if o.ttl > 0 {
  459. ttl -= o.rwait
  460. if ttl <= 0 {
  461. err = ErrTimeout
  462. break
  463. }
  464. resp, err = js.nc.RequestMsg(m, time.Duration(ttl))
  465. } else {
  466. resp, err = js.nc.RequestMsgWithContext(o.ctx, m)
  467. }
  468. }
  469. if err != nil {
  470. if err == ErrNoResponders {
  471. err = ErrNoStreamResponse
  472. }
  473. return nil, err
  474. }
  475. }
  476. var pa pubAckResponse
  477. if err := json.Unmarshal(resp.Data, &pa); err != nil {
  478. return nil, ErrInvalidJSAck
  479. }
  480. if pa.Error != nil {
  481. return nil, pa.Error
  482. }
  483. if pa.PubAck == nil || pa.PubAck.Stream == _EMPTY_ {
  484. return nil, ErrInvalidJSAck
  485. }
  486. return pa.PubAck, nil
  487. }
  488. // Publish publishes a message to a stream from JetStream.
  489. func (js *js) Publish(subj string, data []byte, opts ...PubOpt) (*PubAck, error) {
  490. return js.PublishMsg(&Msg{Subject: subj, Data: data}, opts...)
  491. }
  492. // PubAckFuture is a future for a PubAck.
  493. type PubAckFuture interface {
  494. // Ok returns a receive only channel that can be used to get a PubAck.
  495. Ok() <-chan *PubAck
  496. // Err returns a receive only channel that can be used to get the error from an async publish.
  497. Err() <-chan error
  498. // Msg returns the message that was sent to the server.
  499. Msg() *Msg
  500. }
  501. type pubAckFuture struct {
  502. js *js
  503. msg *Msg
  504. pa *PubAck
  505. st time.Time
  506. err error
  507. errCh chan error
  508. doneCh chan *PubAck
  509. }
  510. func (paf *pubAckFuture) Ok() <-chan *PubAck {
  511. paf.js.mu.Lock()
  512. defer paf.js.mu.Unlock()
  513. if paf.doneCh == nil {
  514. paf.doneCh = make(chan *PubAck, 1)
  515. if paf.pa != nil {
  516. paf.doneCh <- paf.pa
  517. }
  518. }
  519. return paf.doneCh
  520. }
  521. func (paf *pubAckFuture) Err() <-chan error {
  522. paf.js.mu.Lock()
  523. defer paf.js.mu.Unlock()
  524. if paf.errCh == nil {
  525. paf.errCh = make(chan error, 1)
  526. if paf.err != nil {
  527. paf.errCh <- paf.err
  528. }
  529. }
  530. return paf.errCh
  531. }
  532. func (paf *pubAckFuture) Msg() *Msg {
  533. paf.js.mu.RLock()
  534. defer paf.js.mu.RUnlock()
  535. return paf.msg
  536. }
  537. // For quick token lookup etc.
  538. const aReplyPreLen = 14
  539. const aReplyTokensize = 6
  540. func (js *js) newAsyncReply() string {
  541. js.mu.Lock()
  542. if js.rsub == nil {
  543. // Create our wildcard reply subject.
  544. sha := sha256.New()
  545. sha.Write([]byte(nuid.Next()))
  546. b := sha.Sum(nil)
  547. for i := 0; i < aReplyTokensize; i++ {
  548. b[i] = rdigits[int(b[i]%base)]
  549. }
  550. inboxPrefix := InboxPrefix
  551. if js.nc.Opts.InboxPrefix != _EMPTY_ {
  552. inboxPrefix = js.nc.Opts.InboxPrefix + "."
  553. }
  554. js.rpre = fmt.Sprintf("%s%s.", inboxPrefix, b[:aReplyTokensize])
  555. sub, err := js.nc.Subscribe(fmt.Sprintf("%s*", js.rpre), js.handleAsyncReply)
  556. if err != nil {
  557. js.mu.Unlock()
  558. return _EMPTY_
  559. }
  560. js.rsub = sub
  561. js.rr = rand.New(rand.NewSource(time.Now().UnixNano()))
  562. }
  563. if js.connStatusCh == nil {
  564. js.connStatusCh = js.nc.StatusChanged(RECONNECTING, CLOSED)
  565. go js.resetPendingAcksOnReconnect()
  566. }
  567. var sb strings.Builder
  568. sb.WriteString(js.rpre)
  569. rn := js.rr.Int63()
  570. var b [aReplyTokensize]byte
  571. for i, l := 0, rn; i < len(b); i++ {
  572. b[i] = rdigits[l%base]
  573. l /= base
  574. }
  575. sb.Write(b[:])
  576. js.mu.Unlock()
  577. return sb.String()
  578. }
  579. func (js *js) resetPendingAcksOnReconnect() {
  580. js.mu.Lock()
  581. connStatusCh := js.connStatusCh
  582. js.mu.Unlock()
  583. for {
  584. newStatus, ok := <-connStatusCh
  585. if !ok || newStatus == CLOSED {
  586. return
  587. }
  588. js.mu.Lock()
  589. for _, paf := range js.pafs {
  590. paf.err = ErrDisconnected
  591. }
  592. js.pafs = nil
  593. js.mu.Unlock()
  594. }
  595. }
  596. func (js *js) cleanupReplySub() {
  597. js.mu.Lock()
  598. if js.rsub != nil {
  599. js.rsub.Unsubscribe()
  600. js.rsub = nil
  601. }
  602. if js.connStatusCh != nil {
  603. close(js.connStatusCh)
  604. js.connStatusCh = nil
  605. }
  606. js.mu.Unlock()
  607. }
  608. // registerPAF will register for a PubAckFuture.
  609. func (js *js) registerPAF(id string, paf *pubAckFuture) (int, int) {
  610. js.mu.Lock()
  611. if js.pafs == nil {
  612. js.pafs = make(map[string]*pubAckFuture)
  613. }
  614. paf.js = js
  615. js.pafs[id] = paf
  616. np := len(js.pafs)
  617. maxpa := js.opts.maxpa
  618. js.mu.Unlock()
  619. return np, maxpa
  620. }
  621. // Lock should be held.
  622. func (js *js) getPAF(id string) *pubAckFuture {
  623. if js.pafs == nil {
  624. return nil
  625. }
  626. return js.pafs[id]
  627. }
  628. // clearPAF will remove a PubAckFuture that was registered.
  629. func (js *js) clearPAF(id string) {
  630. js.mu.Lock()
  631. delete(js.pafs, id)
  632. js.mu.Unlock()
  633. }
  634. // PublishAsyncPending returns how many PubAckFutures are pending.
  635. func (js *js) PublishAsyncPending() int {
  636. js.mu.RLock()
  637. defer js.mu.RUnlock()
  638. return len(js.pafs)
  639. }
  640. func (js *js) asyncStall() <-chan struct{} {
  641. js.mu.Lock()
  642. if js.stc == nil {
  643. js.stc = make(chan struct{})
  644. }
  645. stc := js.stc
  646. js.mu.Unlock()
  647. return stc
  648. }
  649. // Handle an async reply from PublishAsync.
  650. func (js *js) handleAsyncReply(m *Msg) {
  651. if len(m.Subject) <= aReplyPreLen {
  652. return
  653. }
  654. id := m.Subject[aReplyPreLen:]
  655. js.mu.Lock()
  656. paf := js.getPAF(id)
  657. if paf == nil {
  658. js.mu.Unlock()
  659. return
  660. }
  661. // Remove
  662. delete(js.pafs, id)
  663. // Check on anyone stalled and waiting.
  664. if js.stc != nil && len(js.pafs) < js.opts.maxpa {
  665. close(js.stc)
  666. js.stc = nil
  667. }
  668. // Check on anyone one waiting on done status.
  669. if js.dch != nil && len(js.pafs) == 0 {
  670. dch := js.dch
  671. js.dch = nil
  672. // Defer here so error is processed and can be checked.
  673. defer close(dch)
  674. }
  675. doErr := func(err error) {
  676. paf.err = err
  677. if paf.errCh != nil {
  678. paf.errCh <- paf.err
  679. }
  680. cb := js.opts.aecb
  681. js.mu.Unlock()
  682. if cb != nil {
  683. cb(paf.js, paf.msg, err)
  684. }
  685. }
  686. // Process no responders etc.
  687. if len(m.Data) == 0 && m.Header.Get(statusHdr) == noResponders {
  688. doErr(ErrNoResponders)
  689. return
  690. }
  691. var pa pubAckResponse
  692. if err := json.Unmarshal(m.Data, &pa); err != nil {
  693. doErr(ErrInvalidJSAck)
  694. return
  695. }
  696. if pa.Error != nil {
  697. doErr(pa.Error)
  698. return
  699. }
  700. if pa.PubAck == nil || pa.PubAck.Stream == _EMPTY_ {
  701. doErr(ErrInvalidJSAck)
  702. return
  703. }
  704. // So here we have received a proper puback.
  705. paf.pa = pa.PubAck
  706. if paf.doneCh != nil {
  707. paf.doneCh <- paf.pa
  708. }
  709. js.mu.Unlock()
  710. }
  711. // MsgErrHandler is used to process asynchronous errors from
  712. // JetStream PublishAsync. It will return the original
  713. // message sent to the server for possible retransmitting and the error encountered.
  714. type MsgErrHandler func(JetStream, *Msg, error)
  715. // PublishAsyncErrHandler sets the error handler for async publishes in JetStream.
  716. func PublishAsyncErrHandler(cb MsgErrHandler) JSOpt {
  717. return jsOptFn(func(js *jsOpts) error {
  718. js.aecb = cb
  719. return nil
  720. })
  721. }
  722. // PublishAsyncMaxPending sets the maximum outstanding async publishes that can be inflight at one time.
  723. func PublishAsyncMaxPending(max int) JSOpt {
  724. return jsOptFn(func(js *jsOpts) error {
  725. if max < 1 {
  726. return errors.New("nats: max ack pending should be >= 1")
  727. }
  728. js.maxpa = max
  729. return nil
  730. })
  731. }
  732. // PublishAsync publishes a message to JetStream and returns a PubAckFuture
  733. func (js *js) PublishAsync(subj string, data []byte, opts ...PubOpt) (PubAckFuture, error) {
  734. return js.PublishMsgAsync(&Msg{Subject: subj, Data: data}, opts...)
  735. }
  736. const defaultStallWait = 200 * time.Millisecond
  737. func (js *js) PublishMsgAsync(m *Msg, opts ...PubOpt) (PubAckFuture, error) {
  738. var o pubOpts
  739. if len(opts) > 0 {
  740. if m.Header == nil {
  741. m.Header = Header{}
  742. }
  743. for _, opt := range opts {
  744. if err := opt.configurePublish(&o); err != nil {
  745. return nil, err
  746. }
  747. }
  748. }
  749. // Timeouts and contexts do not make sense for these.
  750. if o.ttl != 0 || o.ctx != nil {
  751. return nil, ErrContextAndTimeout
  752. }
  753. stallWait := defaultStallWait
  754. if o.stallWait > 0 {
  755. stallWait = o.stallWait
  756. }
  757. // FIXME(dlc) - Make common.
  758. if o.id != _EMPTY_ {
  759. m.Header.Set(MsgIdHdr, o.id)
  760. }
  761. if o.lid != _EMPTY_ {
  762. m.Header.Set(ExpectedLastMsgIdHdr, o.lid)
  763. }
  764. if o.str != _EMPTY_ {
  765. m.Header.Set(ExpectedStreamHdr, o.str)
  766. }
  767. if o.seq != nil {
  768. m.Header.Set(ExpectedLastSeqHdr, strconv.FormatUint(*o.seq, 10))
  769. }
  770. if o.lss != nil {
  771. m.Header.Set(ExpectedLastSubjSeqHdr, strconv.FormatUint(*o.lss, 10))
  772. }
  773. // Reply
  774. if m.Reply != _EMPTY_ {
  775. return nil, errors.New("nats: reply subject should be empty")
  776. }
  777. reply := m.Reply
  778. m.Reply = js.newAsyncReply()
  779. defer func() { m.Reply = reply }()
  780. if m.Reply == _EMPTY_ {
  781. return nil, errors.New("nats: error creating async reply handler")
  782. }
  783. id := m.Reply[aReplyPreLen:]
  784. paf := &pubAckFuture{msg: m, st: time.Now()}
  785. numPending, maxPending := js.registerPAF(id, paf)
  786. if maxPending > 0 && numPending >= maxPending {
  787. select {
  788. case <-js.asyncStall():
  789. case <-time.After(stallWait):
  790. js.clearPAF(id)
  791. return nil, errors.New("nats: stalled with too many outstanding async published messages")
  792. }
  793. }
  794. if err := js.nc.PublishMsg(m); err != nil {
  795. js.clearPAF(id)
  796. return nil, err
  797. }
  798. return paf, nil
  799. }
  800. // PublishAsyncComplete returns a channel that will be closed when all outstanding messages have been ack'd.
  801. func (js *js) PublishAsyncComplete() <-chan struct{} {
  802. js.mu.Lock()
  803. defer js.mu.Unlock()
  804. if js.dch == nil {
  805. js.dch = make(chan struct{})
  806. }
  807. dch := js.dch
  808. if len(js.pafs) == 0 {
  809. close(js.dch)
  810. js.dch = nil
  811. }
  812. return dch
  813. }
  814. // MsgId sets the message ID used for deduplication.
  815. func MsgId(id string) PubOpt {
  816. return pubOptFn(func(opts *pubOpts) error {
  817. opts.id = id
  818. return nil
  819. })
  820. }
  821. // ExpectStream sets the expected stream to respond from the publish.
  822. func ExpectStream(stream string) PubOpt {
  823. return pubOptFn(func(opts *pubOpts) error {
  824. opts.str = stream
  825. return nil
  826. })
  827. }
  828. // ExpectLastSequence sets the expected sequence in the response from the publish.
  829. func ExpectLastSequence(seq uint64) PubOpt {
  830. return pubOptFn(func(opts *pubOpts) error {
  831. opts.seq = &seq
  832. return nil
  833. })
  834. }
  835. // ExpectLastSequencePerSubject sets the expected sequence per subject in the response from the publish.
  836. func ExpectLastSequencePerSubject(seq uint64) PubOpt {
  837. return pubOptFn(func(opts *pubOpts) error {
  838. opts.lss = &seq
  839. return nil
  840. })
  841. }
  842. // ExpectLastMsgId sets the expected last msgId in the response from the publish.
  843. func ExpectLastMsgId(id string) PubOpt {
  844. return pubOptFn(func(opts *pubOpts) error {
  845. opts.lid = id
  846. return nil
  847. })
  848. }
  849. // RetryWait sets the retry wait time when ErrNoResponders is encountered.
  850. func RetryWait(dur time.Duration) PubOpt {
  851. return pubOptFn(func(opts *pubOpts) error {
  852. opts.rwait = dur
  853. return nil
  854. })
  855. }
  856. // RetryAttempts sets the retry number of attempts when ErrNoResponders is encountered.
  857. func RetryAttempts(num int) PubOpt {
  858. return pubOptFn(func(opts *pubOpts) error {
  859. opts.rnum = num
  860. return nil
  861. })
  862. }
  863. // StallWait sets the max wait when the producer becomes stall producing messages.
  864. func StallWait(ttl time.Duration) PubOpt {
  865. return pubOptFn(func(opts *pubOpts) error {
  866. if ttl <= 0 {
  867. return fmt.Errorf("nats: stall wait should be more than 0")
  868. }
  869. opts.stallWait = ttl
  870. return nil
  871. })
  872. }
  873. type ackOpts struct {
  874. ttl time.Duration
  875. ctx context.Context
  876. nakDelay time.Duration
  877. }
  878. // AckOpt are the options that can be passed when acknowledge a message.
  879. type AckOpt interface {
  880. configureAck(opts *ackOpts) error
  881. }
  882. // MaxWait sets the maximum amount of time we will wait for a response.
  883. type MaxWait time.Duration
  884. func (ttl MaxWait) configureJSContext(js *jsOpts) error {
  885. js.wait = time.Duration(ttl)
  886. return nil
  887. }
  888. func (ttl MaxWait) configurePull(opts *pullOpts) error {
  889. opts.ttl = time.Duration(ttl)
  890. return nil
  891. }
  892. // AckWait sets the maximum amount of time we will wait for an ack.
  893. type AckWait time.Duration
  894. func (ttl AckWait) configurePublish(opts *pubOpts) error {
  895. opts.ttl = time.Duration(ttl)
  896. return nil
  897. }
  898. func (ttl AckWait) configureSubscribe(opts *subOpts) error {
  899. opts.cfg.AckWait = time.Duration(ttl)
  900. return nil
  901. }
  902. func (ttl AckWait) configureAck(opts *ackOpts) error {
  903. opts.ttl = time.Duration(ttl)
  904. return nil
  905. }
  906. // ContextOpt is an option used to set a context.Context.
  907. type ContextOpt struct {
  908. context.Context
  909. }
  910. func (ctx ContextOpt) configureJSContext(opts *jsOpts) error {
  911. opts.ctx = ctx
  912. return nil
  913. }
  914. func (ctx ContextOpt) configurePublish(opts *pubOpts) error {
  915. opts.ctx = ctx
  916. return nil
  917. }
  918. func (ctx ContextOpt) configureSubscribe(opts *subOpts) error {
  919. opts.ctx = ctx
  920. return nil
  921. }
  922. func (ctx ContextOpt) configurePull(opts *pullOpts) error {
  923. opts.ctx = ctx
  924. return nil
  925. }
  926. func (ctx ContextOpt) configureAck(opts *ackOpts) error {
  927. opts.ctx = ctx
  928. return nil
  929. }
  930. // Context returns an option that can be used to configure a context for APIs
  931. // that are context aware such as those part of the JetStream interface.
  932. func Context(ctx context.Context) ContextOpt {
  933. return ContextOpt{ctx}
  934. }
  935. type nakDelay time.Duration
  936. func (d nakDelay) configureAck(opts *ackOpts) error {
  937. opts.nakDelay = time.Duration(d)
  938. return nil
  939. }
  940. // Subscribe
  941. // ConsumerConfig is the configuration of a JetStream consumer.
  942. type ConsumerConfig struct {
  943. Durable string `json:"durable_name,omitempty"`
  944. Name string `json:"name,omitempty"`
  945. Description string `json:"description,omitempty"`
  946. DeliverPolicy DeliverPolicy `json:"deliver_policy"`
  947. OptStartSeq uint64 `json:"opt_start_seq,omitempty"`
  948. OptStartTime *time.Time `json:"opt_start_time,omitempty"`
  949. AckPolicy AckPolicy `json:"ack_policy"`
  950. AckWait time.Duration `json:"ack_wait,omitempty"`
  951. MaxDeliver int `json:"max_deliver,omitempty"`
  952. BackOff []time.Duration `json:"backoff,omitempty"`
  953. FilterSubject string `json:"filter_subject,omitempty"`
  954. ReplayPolicy ReplayPolicy `json:"replay_policy"`
  955. RateLimit uint64 `json:"rate_limit_bps,omitempty"` // Bits per sec
  956. SampleFrequency string `json:"sample_freq,omitempty"`
  957. MaxWaiting int `json:"max_waiting,omitempty"`
  958. MaxAckPending int `json:"max_ack_pending,omitempty"`
  959. FlowControl bool `json:"flow_control,omitempty"`
  960. Heartbeat time.Duration `json:"idle_heartbeat,omitempty"`
  961. HeadersOnly bool `json:"headers_only,omitempty"`
  962. // Pull based options.
  963. MaxRequestBatch int `json:"max_batch,omitempty"`
  964. MaxRequestExpires time.Duration `json:"max_expires,omitempty"`
  965. MaxRequestMaxBytes int `json:"max_bytes,omitempty"`
  966. // Push based consumers.
  967. DeliverSubject string `json:"deliver_subject,omitempty"`
  968. DeliverGroup string `json:"deliver_group,omitempty"`
  969. // Inactivity threshold.
  970. InactiveThreshold time.Duration `json:"inactive_threshold,omitempty"`
  971. // Generally inherited by parent stream and other markers, now can be configured directly.
  972. Replicas int `json:"num_replicas"`
  973. // Force memory storage.
  974. MemoryStorage bool `json:"mem_storage,omitempty"`
  975. }
  976. // ConsumerInfo is the info from a JetStream consumer.
  977. type ConsumerInfo struct {
  978. Stream string `json:"stream_name"`
  979. Name string `json:"name"`
  980. Created time.Time `json:"created"`
  981. Config ConsumerConfig `json:"config"`
  982. Delivered SequenceInfo `json:"delivered"`
  983. AckFloor SequenceInfo `json:"ack_floor"`
  984. NumAckPending int `json:"num_ack_pending"`
  985. NumRedelivered int `json:"num_redelivered"`
  986. NumWaiting int `json:"num_waiting"`
  987. NumPending uint64 `json:"num_pending"`
  988. Cluster *ClusterInfo `json:"cluster,omitempty"`
  989. PushBound bool `json:"push_bound,omitempty"`
  990. }
  991. // SequenceInfo has both the consumer and the stream sequence and last activity.
  992. type SequenceInfo struct {
  993. Consumer uint64 `json:"consumer_seq"`
  994. Stream uint64 `json:"stream_seq"`
  995. Last *time.Time `json:"last_active,omitempty"`
  996. }
  997. // SequencePair includes the consumer and stream sequence info from a JetStream consumer.
  998. type SequencePair struct {
  999. Consumer uint64 `json:"consumer_seq"`
  1000. Stream uint64 `json:"stream_seq"`
  1001. }
  1002. // nextRequest is for getting next messages for pull based consumers from JetStream.
  1003. type nextRequest struct {
  1004. Expires time.Duration `json:"expires,omitempty"`
  1005. Batch int `json:"batch,omitempty"`
  1006. NoWait bool `json:"no_wait,omitempty"`
  1007. MaxBytes int `json:"max_bytes,omitempty"`
  1008. }
  1009. // jsSub includes JetStream subscription info.
  1010. type jsSub struct {
  1011. js *js
  1012. // For pull subscribers, this is the next message subject to send requests to.
  1013. nms string
  1014. psubj string // the subject that was passed by user to the subscribe calls
  1015. consumer string
  1016. stream string
  1017. deliver string
  1018. pull bool
  1019. dc bool // Delete JS consumer
  1020. ackNone bool
  1021. // This is ConsumerInfo's Pending+Consumer.Delivered that we get from the
  1022. // add consumer response. Note that some versions of the server gather the
  1023. // consumer info *after* the creation of the consumer, which means that
  1024. // some messages may have been already delivered. So the sum of the two
  1025. // is a more accurate representation of the number of messages pending or
  1026. // in the process of being delivered to the subscription when created.
  1027. pending uint64
  1028. // Ordered consumers
  1029. ordered bool
  1030. dseq uint64
  1031. sseq uint64
  1032. ccreq *createConsumerRequest
  1033. // Heartbeats and Flow Control handling from push consumers.
  1034. hbc *time.Timer
  1035. hbi time.Duration
  1036. active bool
  1037. cmeta string
  1038. fcr string
  1039. fcd uint64
  1040. fciseq uint64
  1041. csfct *time.Timer
  1042. // Cancellation function to cancel context on drain/unsubscribe.
  1043. cancel func()
  1044. }
  1045. // Deletes the JS Consumer.
  1046. // No connection nor subscription lock must be held on entry.
  1047. func (sub *Subscription) deleteConsumer() error {
  1048. sub.mu.Lock()
  1049. jsi := sub.jsi
  1050. if jsi == nil {
  1051. sub.mu.Unlock()
  1052. return nil
  1053. }
  1054. stream, consumer := jsi.stream, jsi.consumer
  1055. js := jsi.js
  1056. sub.mu.Unlock()
  1057. return js.DeleteConsumer(stream, consumer)
  1058. }
  1059. // SubOpt configures options for subscribing to JetStream consumers.
  1060. type SubOpt interface {
  1061. configureSubscribe(opts *subOpts) error
  1062. }
  1063. // subOptFn is a function option used to configure a JetStream Subscribe.
  1064. type subOptFn func(opts *subOpts) error
  1065. func (opt subOptFn) configureSubscribe(opts *subOpts) error {
  1066. return opt(opts)
  1067. }
  1068. // Subscribe creates an async Subscription for JetStream.
  1069. // The stream and consumer names can be provided with the nats.Bind() option.
  1070. // For creating an ephemeral (where the consumer name is picked by the server),
  1071. // you can provide the stream name with nats.BindStream().
  1072. // If no stream name is specified, the library will attempt to figure out which
  1073. // stream the subscription is for. See important notes below for more details.
  1074. //
  1075. // IMPORTANT NOTES:
  1076. // * If none of the options Bind() nor Durable() are specified, the library will
  1077. // send a request to the server to create an ephemeral JetStream consumer,
  1078. // which will be deleted after an Unsubscribe() or Drain(), or automatically
  1079. // by the server after a short period of time after the NATS subscription is
  1080. // gone.
  1081. // * If Durable() option is specified, the library will attempt to lookup a JetStream
  1082. // consumer with this name, and if found, will bind to it and not attempt to
  1083. // delete it. However, if not found, the library will send a request to create
  1084. // such durable JetStream consumer. The library will delete the JetStream consumer
  1085. // after an Unsubscribe() or Drain().
  1086. // * If Bind() option is provided, the library will attempt to lookup the
  1087. // consumer with the given name, and if successful, bind to it. If the lookup fails,
  1088. // then the Subscribe() call will return an error.
  1089. func (js *js) Subscribe(subj string, cb MsgHandler, opts ...SubOpt) (*Subscription, error) {
  1090. if cb == nil {
  1091. return nil, ErrBadSubscription
  1092. }
  1093. return js.subscribe(subj, _EMPTY_, cb, nil, false, false, opts)
  1094. }
  1095. // SubscribeSync creates a Subscription that can be used to process messages synchronously.
  1096. // See important note in Subscribe()
  1097. func (js *js) SubscribeSync(subj string, opts ...SubOpt) (*Subscription, error) {
  1098. mch := make(chan *Msg, js.nc.Opts.SubChanLen)
  1099. return js.subscribe(subj, _EMPTY_, nil, mch, true, false, opts)
  1100. }
  1101. // QueueSubscribe creates a Subscription with a queue group.
  1102. // If no optional durable name nor binding options are specified, the queue name will be used as a durable name.
  1103. // See important note in Subscribe()
  1104. func (js *js) QueueSubscribe(subj, queue string, cb MsgHandler, opts ...SubOpt) (*Subscription, error) {
  1105. if cb == nil {
  1106. return nil, ErrBadSubscription
  1107. }
  1108. return js.subscribe(subj, queue, cb, nil, false, false, opts)
  1109. }
  1110. // QueueSubscribeSync creates a Subscription with a queue group that can be used to process messages synchronously.
  1111. // See important note in QueueSubscribe()
  1112. func (js *js) QueueSubscribeSync(subj, queue string, opts ...SubOpt) (*Subscription, error) {
  1113. mch := make(chan *Msg, js.nc.Opts.SubChanLen)
  1114. return js.subscribe(subj, queue, nil, mch, true, false, opts)
  1115. }
  1116. // ChanSubscribe creates channel based Subscription.
  1117. // Using ChanSubscribe without buffered capacity is not recommended since
  1118. // it will be prone to dropping messages with a slow consumer error. Make sure to give the channel enough
  1119. // capacity to handle bursts in traffic, for example other Subscribe APIs use a default of 512k capacity in comparison.
  1120. // See important note in Subscribe()
  1121. func (js *js) ChanSubscribe(subj string, ch chan *Msg, opts ...SubOpt) (*Subscription, error) {
  1122. return js.subscribe(subj, _EMPTY_, nil, ch, false, false, opts)
  1123. }
  1124. // ChanQueueSubscribe creates channel based Subscription with a queue group.
  1125. // See important note in QueueSubscribe()
  1126. func (js *js) ChanQueueSubscribe(subj, queue string, ch chan *Msg, opts ...SubOpt) (*Subscription, error) {
  1127. return js.subscribe(subj, queue, nil, ch, false, false, opts)
  1128. }
  1129. // PullSubscribe creates a Subscription that can fetch messages.
  1130. // See important note in Subscribe()
  1131. func (js *js) PullSubscribe(subj, durable string, opts ...SubOpt) (*Subscription, error) {
  1132. mch := make(chan *Msg, js.nc.Opts.SubChanLen)
  1133. if durable != "" {
  1134. opts = append(opts, Durable(durable))
  1135. }
  1136. return js.subscribe(subj, _EMPTY_, nil, mch, true, true, opts)
  1137. }
  1138. func processConsInfo(info *ConsumerInfo, userCfg *ConsumerConfig, isPullMode bool, subj, queue string) (string, error) {
  1139. ccfg := &info.Config
  1140. // Make sure this new subject matches or is a subset.
  1141. if ccfg.FilterSubject != _EMPTY_ && subj != ccfg.FilterSubject {
  1142. return _EMPTY_, ErrSubjectMismatch
  1143. }
  1144. // Prevent binding a subscription against incompatible consumer types.
  1145. if isPullMode && ccfg.DeliverSubject != _EMPTY_ {
  1146. return _EMPTY_, ErrPullSubscribeToPushConsumer
  1147. } else if !isPullMode && ccfg.DeliverSubject == _EMPTY_ {
  1148. return _EMPTY_, ErrPullSubscribeRequired
  1149. }
  1150. // If pull mode, nothing else to check here.
  1151. if isPullMode {
  1152. return _EMPTY_, checkConfig(ccfg, userCfg)
  1153. }
  1154. // At this point, we know the user wants push mode, and the JS consumer is
  1155. // really push mode.
  1156. dg := info.Config.DeliverGroup
  1157. if dg == _EMPTY_ {
  1158. // Prevent an user from attempting to create a queue subscription on
  1159. // a JS consumer that was not created with a deliver group.
  1160. if queue != _EMPTY_ {
  1161. return _EMPTY_, fmt.Errorf("cannot create a queue subscription for a consumer without a deliver group")
  1162. } else if info.PushBound {
  1163. // Need to reject a non queue subscription to a non queue consumer
  1164. // if the consumer is already bound.
  1165. return _EMPTY_, fmt.Errorf("consumer is already bound to a subscription")
  1166. }
  1167. } else {
  1168. // If the JS consumer has a deliver group, we need to fail a non queue
  1169. // subscription attempt:
  1170. if queue == _EMPTY_ {
  1171. return _EMPTY_, fmt.Errorf("cannot create a subscription for a consumer with a deliver group %q", dg)
  1172. } else if queue != dg {
  1173. // Here the user's queue group name does not match the one associated
  1174. // with the JS consumer.
  1175. return _EMPTY_, fmt.Errorf("cannot create a queue subscription %q for a consumer with a deliver group %q",
  1176. queue, dg)
  1177. }
  1178. }
  1179. if err := checkConfig(ccfg, userCfg); err != nil {
  1180. return _EMPTY_, err
  1181. }
  1182. return ccfg.DeliverSubject, nil
  1183. }
  1184. func checkConfig(s, u *ConsumerConfig) error {
  1185. makeErr := func(fieldName string, usrVal, srvVal any) error {
  1186. return fmt.Errorf("configuration requests %s to be %v, but consumer's value is %v", fieldName, usrVal, srvVal)
  1187. }
  1188. if u.Durable != _EMPTY_ && u.Durable != s.Durable {
  1189. return makeErr("durable", u.Durable, s.Durable)
  1190. }
  1191. if u.Description != _EMPTY_ && u.Description != s.Description {
  1192. return makeErr("description", u.Description, s.Description)
  1193. }
  1194. if u.DeliverPolicy != deliverPolicyNotSet && u.DeliverPolicy != s.DeliverPolicy {
  1195. return makeErr("deliver policy", u.DeliverPolicy, s.DeliverPolicy)
  1196. }
  1197. if u.OptStartSeq > 0 && u.OptStartSeq != s.OptStartSeq {
  1198. return makeErr("optional start sequence", u.OptStartSeq, s.OptStartSeq)
  1199. }
  1200. if u.OptStartTime != nil && !u.OptStartTime.IsZero() && !(*u.OptStartTime).Equal(*s.OptStartTime) {
  1201. return makeErr("optional start time", u.OptStartTime, s.OptStartTime)
  1202. }
  1203. if u.AckPolicy != ackPolicyNotSet && u.AckPolicy != s.AckPolicy {
  1204. return makeErr("ack policy", u.AckPolicy, s.AckPolicy)
  1205. }
  1206. if u.AckWait > 0 && u.AckWait != s.AckWait {
  1207. return makeErr("ack wait", u.AckWait, s.AckWait)
  1208. }
  1209. if u.MaxDeliver > 0 && u.MaxDeliver != s.MaxDeliver {
  1210. return makeErr("max deliver", u.MaxDeliver, s.MaxDeliver)
  1211. }
  1212. if u.ReplayPolicy != replayPolicyNotSet && u.ReplayPolicy != s.ReplayPolicy {
  1213. return makeErr("replay policy", u.ReplayPolicy, s.ReplayPolicy)
  1214. }
  1215. if u.RateLimit > 0 && u.RateLimit != s.RateLimit {
  1216. return makeErr("rate limit", u.RateLimit, s.RateLimit)
  1217. }
  1218. if u.SampleFrequency != _EMPTY_ && u.SampleFrequency != s.SampleFrequency {
  1219. return makeErr("sample frequency", u.SampleFrequency, s.SampleFrequency)
  1220. }
  1221. if u.MaxWaiting > 0 && u.MaxWaiting != s.MaxWaiting {
  1222. return makeErr("max waiting", u.MaxWaiting, s.MaxWaiting)
  1223. }
  1224. if u.MaxAckPending > 0 && u.MaxAckPending != s.MaxAckPending {
  1225. return makeErr("max ack pending", u.MaxAckPending, s.MaxAckPending)
  1226. }
  1227. // For flow control, we want to fail if the user explicit wanted it, but
  1228. // it is not set in the existing consumer. If it is not asked by the user,
  1229. // the library still handles it and so no reason to fail.
  1230. if u.FlowControl && !s.FlowControl {
  1231. return makeErr("flow control", u.FlowControl, s.FlowControl)
  1232. }
  1233. if u.Heartbeat > 0 && u.Heartbeat != s.Heartbeat {
  1234. return makeErr("heartbeat", u.Heartbeat, s.Heartbeat)
  1235. }
  1236. if u.Replicas > 0 && u.Replicas != s.Replicas {
  1237. return makeErr("replicas", u.Replicas, s.Replicas)
  1238. }
  1239. if u.MemoryStorage && !s.MemoryStorage {
  1240. return makeErr("memory storage", u.MemoryStorage, s.MemoryStorage)
  1241. }
  1242. return nil
  1243. }
  1244. func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync, isPullMode bool, opts []SubOpt) (*Subscription, error) {
  1245. cfg := ConsumerConfig{
  1246. DeliverPolicy: deliverPolicyNotSet,
  1247. AckPolicy: ackPolicyNotSet,
  1248. ReplayPolicy: replayPolicyNotSet,
  1249. }
  1250. o := subOpts{cfg: &cfg}
  1251. if len(opts) > 0 {
  1252. for _, opt := range opts {
  1253. if opt == nil {
  1254. continue
  1255. }
  1256. if err := opt.configureSubscribe(&o); err != nil {
  1257. return nil, err
  1258. }
  1259. }
  1260. }
  1261. // If no stream name is specified, the subject cannot be empty.
  1262. if subj == _EMPTY_ && o.stream == _EMPTY_ {
  1263. return nil, fmt.Errorf("nats: subject required")
  1264. }
  1265. // Note that these may change based on the consumer info response we may get.
  1266. hasHeartbeats := o.cfg.Heartbeat > 0
  1267. hasFC := o.cfg.FlowControl
  1268. // Some checks for pull subscribers
  1269. if isPullMode {
  1270. // No deliver subject should be provided
  1271. if o.cfg.DeliverSubject != _EMPTY_ {
  1272. return nil, ErrPullSubscribeToPushConsumer
  1273. }
  1274. }
  1275. // Some check/setting specific to queue subs
  1276. if queue != _EMPTY_ {
  1277. // Queue subscriber cannot have HB or FC (since messages will be randomly dispatched
  1278. // to members). We may in the future have a separate NATS subscription that all members
  1279. // would subscribe to and server would send on.
  1280. if o.cfg.Heartbeat > 0 || o.cfg.FlowControl {
  1281. // Not making this a public ErrXXX in case we allow in the future.
  1282. return nil, fmt.Errorf("nats: queue subscription doesn't support idle heartbeat nor flow control")
  1283. }
  1284. // If this is a queue subscription and no consumer nor durable name was specified,
  1285. // then we will use the queue name as a durable name.
  1286. if o.consumer == _EMPTY_ && o.cfg.Durable == _EMPTY_ {
  1287. if err := checkConsumerName(queue); err != nil {
  1288. return nil, err
  1289. }
  1290. o.cfg.Durable = queue
  1291. }
  1292. }
  1293. var (
  1294. err error
  1295. shouldCreate bool
  1296. info *ConsumerInfo
  1297. deliver string
  1298. stream = o.stream
  1299. consumer = o.consumer
  1300. isDurable = o.cfg.Durable != _EMPTY_
  1301. consumerBound = o.bound
  1302. ctx = o.ctx
  1303. skipCInfo = o.skipCInfo
  1304. notFoundErr bool
  1305. lookupErr bool
  1306. nc = js.nc
  1307. nms string
  1308. hbi time.Duration
  1309. ccreq *createConsumerRequest // In case we need to hold onto it for ordered consumers.
  1310. maxap int
  1311. )
  1312. // Do some quick checks here for ordered consumers. We do these here instead of spread out
  1313. // in the individual SubOpts.
  1314. if o.ordered {
  1315. // Make sure we are not durable.
  1316. if isDurable {
  1317. return nil, fmt.Errorf("nats: durable can not be set for an ordered consumer")
  1318. }
  1319. // Check ack policy.
  1320. if o.cfg.AckPolicy != ackPolicyNotSet {
  1321. return nil, fmt.Errorf("nats: ack policy can not be set for an ordered consumer")
  1322. }
  1323. // Check max deliver.
  1324. if o.cfg.MaxDeliver != 1 && o.cfg.MaxDeliver != 0 {
  1325. return nil, fmt.Errorf("nats: max deliver can not be set for an ordered consumer")
  1326. }
  1327. // No deliver subject, we pick our own.
  1328. if o.cfg.DeliverSubject != _EMPTY_ {
  1329. return nil, fmt.Errorf("nats: deliver subject can not be set for an ordered consumer")
  1330. }
  1331. // Queue groups not allowed.
  1332. if queue != _EMPTY_ {
  1333. return nil, fmt.Errorf("nats: queues not be set for an ordered consumer")
  1334. }
  1335. // Check for bound consumers.
  1336. if consumer != _EMPTY_ {
  1337. return nil, fmt.Errorf("nats: can not bind existing consumer for an ordered consumer")
  1338. }
  1339. // Check for pull mode.
  1340. if isPullMode {
  1341. return nil, fmt.Errorf("nats: can not use pull mode for an ordered consumer")
  1342. }
  1343. // Setup how we need it to be here.
  1344. o.cfg.FlowControl = true
  1345. o.cfg.AckPolicy = AckNonePolicy
  1346. o.cfg.MaxDeliver = 1
  1347. o.cfg.AckWait = 22 * time.Hour // Just set to something known, not utilized.
  1348. // Force R1 and MemoryStorage for these.
  1349. o.cfg.Replicas = 1
  1350. o.cfg.MemoryStorage = true
  1351. if !hasHeartbeats {
  1352. o.cfg.Heartbeat = orderedHeartbeatsInterval
  1353. }
  1354. hasFC, hasHeartbeats = true, true
  1355. o.mack = true // To avoid auto-ack wrapping call below.
  1356. hbi = o.cfg.Heartbeat
  1357. }
  1358. // In case a consumer has not been set explicitly, then the
  1359. // durable name will be used as the consumer name.
  1360. if consumer == _EMPTY_ {
  1361. consumer = o.cfg.Durable
  1362. }
  1363. // Find the stream mapped to the subject if not bound to a stream already.
  1364. if stream == _EMPTY_ {
  1365. stream, err = js.StreamNameBySubject(subj)
  1366. if err != nil {
  1367. return nil, err
  1368. }
  1369. }
  1370. // With an explicit durable name, we can lookup the consumer first
  1371. // to which it should be attaching to.
  1372. // If SkipConsumerLookup was used, do not call consumer info.
  1373. if consumer != _EMPTY_ && !o.skipCInfo {
  1374. info, err = js.ConsumerInfo(stream, consumer)
  1375. notFoundErr = errors.Is(err, ErrConsumerNotFound)
  1376. lookupErr = err == ErrJetStreamNotEnabled || err == ErrTimeout || err == context.DeadlineExceeded
  1377. }
  1378. switch {
  1379. case info != nil:
  1380. deliver, err = processConsInfo(info, o.cfg, isPullMode, subj, queue)
  1381. if err != nil {
  1382. return nil, err
  1383. }
  1384. icfg := &info.Config
  1385. hasFC, hbi = icfg.FlowControl, icfg.Heartbeat
  1386. hasHeartbeats = hbi > 0
  1387. maxap = icfg.MaxAckPending
  1388. case (err != nil && !notFoundErr) || (notFoundErr && consumerBound):
  1389. // If the consumer is being bound and we got an error on pull subscribe then allow the error.
  1390. if !(isPullMode && lookupErr && consumerBound) {
  1391. return nil, err
  1392. }
  1393. case skipCInfo:
  1394. // When skipping consumer info, need to rely on the manually passed sub options
  1395. // to match the expected behavior from the subscription.
  1396. hasFC, hbi = o.cfg.FlowControl, o.cfg.Heartbeat
  1397. hasHeartbeats = hbi > 0
  1398. maxap = o.cfg.MaxAckPending
  1399. deliver = o.cfg.DeliverSubject
  1400. if consumerBound {
  1401. break
  1402. }
  1403. // When not bound to a consumer already, proceed to create.
  1404. fallthrough
  1405. default:
  1406. // Attempt to create consumer if not found nor using Bind.
  1407. shouldCreate = true
  1408. if o.cfg.DeliverSubject != _EMPTY_ {
  1409. deliver = o.cfg.DeliverSubject
  1410. } else if !isPullMode {
  1411. deliver = nc.NewInbox()
  1412. cfg.DeliverSubject = deliver
  1413. }
  1414. // Do filtering always, server will clear as needed.
  1415. cfg.FilterSubject = subj
  1416. // Pass the queue to the consumer config
  1417. if queue != _EMPTY_ {
  1418. cfg.DeliverGroup = queue
  1419. }
  1420. // If not set, default to deliver all
  1421. if cfg.DeliverPolicy == deliverPolicyNotSet {
  1422. cfg.DeliverPolicy = DeliverAllPolicy
  1423. }
  1424. // If not set, default to ack explicit.
  1425. if cfg.AckPolicy == ackPolicyNotSet {
  1426. cfg.AckPolicy = AckExplicitPolicy
  1427. }
  1428. // If not set, default to instant
  1429. if cfg.ReplayPolicy == replayPolicyNotSet {
  1430. cfg.ReplayPolicy = ReplayInstantPolicy
  1431. }
  1432. // If we have acks at all and the MaxAckPending is not set go ahead
  1433. // and set to the internal max for channel based consumers
  1434. if cfg.MaxAckPending == 0 && ch != nil && cfg.AckPolicy != AckNonePolicy {
  1435. cfg.MaxAckPending = cap(ch)
  1436. }
  1437. // Create request here.
  1438. ccreq = &createConsumerRequest{
  1439. Stream: stream,
  1440. Config: &cfg,
  1441. }
  1442. hbi = cfg.Heartbeat
  1443. }
  1444. if isPullMode {
  1445. nms = fmt.Sprintf(js.apiSubj(apiRequestNextT), stream, consumer)
  1446. deliver = nc.NewInbox()
  1447. // for pull consumers, create a wildcard subscription to differentiate pull requests
  1448. deliver += ".*"
  1449. }
  1450. // In case this has a context, then create a child context that
  1451. // is possible to cancel via unsubscribe / drain.
  1452. var cancel func()
  1453. if ctx != nil {
  1454. ctx, cancel = context.WithCancel(ctx)
  1455. }
  1456. jsi := &jsSub{
  1457. js: js,
  1458. stream: stream,
  1459. consumer: consumer,
  1460. deliver: deliver,
  1461. hbi: hbi,
  1462. ordered: o.ordered,
  1463. ccreq: ccreq,
  1464. dseq: 1,
  1465. pull: isPullMode,
  1466. nms: nms,
  1467. psubj: subj,
  1468. cancel: cancel,
  1469. ackNone: o.cfg.AckPolicy == AckNonePolicy,
  1470. }
  1471. // Auto acknowledge unless manual ack is set or policy is set to AckNonePolicy
  1472. if cb != nil && !o.mack && o.cfg.AckPolicy != AckNonePolicy {
  1473. ocb := cb
  1474. cb = func(m *Msg) { ocb(m); m.Ack() }
  1475. }
  1476. sub, err := nc.subscribe(deliver, queue, cb, ch, isSync, jsi)
  1477. if err != nil {
  1478. return nil, err
  1479. }
  1480. // If we fail and we had the sub we need to cleanup, but can't just do a straight Unsubscribe or Drain.
  1481. // We need to clear the jsi so we do not remove any durables etc.
  1482. cleanUpSub := func() {
  1483. if sub != nil {
  1484. sub.mu.Lock()
  1485. sub.jsi = nil
  1486. sub.mu.Unlock()
  1487. sub.Unsubscribe()
  1488. }
  1489. }
  1490. // If we are creating or updating let's process that request.
  1491. consName := o.cfg.Name
  1492. if shouldCreate {
  1493. if cfg.Durable != "" {
  1494. consName = cfg.Durable
  1495. } else if consName == "" {
  1496. consName = getHash(nuid.Next())
  1497. }
  1498. info, err := js.upsertConsumer(stream, consName, ccreq.Config)
  1499. if err != nil {
  1500. var apiErr *APIError
  1501. if ok := errors.As(err, &apiErr); !ok {
  1502. cleanUpSub()
  1503. return nil, err
  1504. }
  1505. if consumer == _EMPTY_ ||
  1506. (apiErr.ErrorCode != JSErrCodeConsumerAlreadyExists && apiErr.ErrorCode != JSErrCodeConsumerNameExists) {
  1507. cleanUpSub()
  1508. if errors.Is(apiErr, ErrStreamNotFound) {
  1509. return nil, ErrStreamNotFound
  1510. }
  1511. return nil, err
  1512. }
  1513. // We will not be using this sub here if we were push based.
  1514. if !isPullMode {
  1515. cleanUpSub()
  1516. }
  1517. info, err = js.ConsumerInfo(stream, consumer)
  1518. if err != nil {
  1519. return nil, err
  1520. }
  1521. deliver, err = processConsInfo(info, o.cfg, isPullMode, subj, queue)
  1522. if err != nil {
  1523. return nil, err
  1524. }
  1525. if !isPullMode {
  1526. // We can't reuse the channel, so if one was passed, we need to create a new one.
  1527. if isSync {
  1528. ch = make(chan *Msg, cap(ch))
  1529. } else if ch != nil {
  1530. // User provided (ChanSubscription), simply try to drain it.
  1531. for done := false; !done; {
  1532. select {
  1533. case <-ch:
  1534. default:
  1535. done = true
  1536. }
  1537. }
  1538. }
  1539. jsi.deliver = deliver
  1540. jsi.hbi = info.Config.Heartbeat
  1541. // Recreate the subscription here.
  1542. sub, err = nc.subscribe(jsi.deliver, queue, cb, ch, isSync, jsi)
  1543. if err != nil {
  1544. return nil, err
  1545. }
  1546. hasFC = info.Config.FlowControl
  1547. hasHeartbeats = info.Config.Heartbeat > 0
  1548. }
  1549. } else {
  1550. // Since the library created the JS consumer, it will delete it on Unsubscribe()/Drain()
  1551. sub.mu.Lock()
  1552. sub.jsi.dc = true
  1553. sub.jsi.pending = info.NumPending + info.Delivered.Consumer
  1554. // If this is an ephemeral, we did not have a consumer name, we get it from the info
  1555. // after the AddConsumer returns.
  1556. if consumer == _EMPTY_ {
  1557. sub.jsi.consumer = info.Name
  1558. if isPullMode {
  1559. sub.jsi.nms = fmt.Sprintf(js.apiSubj(apiRequestNextT), stream, info.Name)
  1560. }
  1561. }
  1562. sub.mu.Unlock()
  1563. }
  1564. // Capture max ack pending from the info response here which covers both
  1565. // success and failure followed by consumer lookup.
  1566. maxap = info.Config.MaxAckPending
  1567. }
  1568. // If maxap is greater than the default sub's pending limit, use that.
  1569. if maxap > DefaultSubPendingMsgsLimit {
  1570. // For bytes limit, use the min of maxp*1MB or DefaultSubPendingBytesLimit
  1571. bl := maxap * 1024 * 1024
  1572. if bl < DefaultSubPendingBytesLimit {
  1573. bl = DefaultSubPendingBytesLimit
  1574. }
  1575. sub.SetPendingLimits(maxap, bl)
  1576. }
  1577. // Do heartbeats last if needed.
  1578. if hasHeartbeats {
  1579. sub.scheduleHeartbeatCheck()
  1580. }
  1581. // For ChanSubscriptions, if we know that there is flow control, we will
  1582. // start a go routine that evaluates the number of delivered messages
  1583. // and process flow control.
  1584. if sub.Type() == ChanSubscription && hasFC {
  1585. sub.chanSubcheckForFlowControlResponse()
  1586. }
  1587. // Wait for context to get canceled if there is one.
  1588. if ctx != nil {
  1589. go func() {
  1590. <-ctx.Done()
  1591. sub.Unsubscribe()
  1592. }()
  1593. }
  1594. return sub, nil
  1595. }
  1596. // This long-lived routine is used per ChanSubscription to check
  1597. // on the number of delivered messages and check for flow control response.
  1598. func (sub *Subscription) chanSubcheckForFlowControlResponse() {
  1599. sub.mu.Lock()
  1600. // We don't use defer since if we need to send an RC reply, we need
  1601. // to do it outside the sub's lock. So doing explicit unlock...
  1602. if sub.closed {
  1603. sub.mu.Unlock()
  1604. return
  1605. }
  1606. var fcReply string
  1607. var nc *Conn
  1608. jsi := sub.jsi
  1609. if jsi.csfct == nil {
  1610. jsi.csfct = time.AfterFunc(chanSubFCCheckInterval, sub.chanSubcheckForFlowControlResponse)
  1611. } else {
  1612. fcReply = sub.checkForFlowControlResponse()
  1613. nc = sub.conn
  1614. // Do the reset here under the lock, it's ok...
  1615. jsi.csfct.Reset(chanSubFCCheckInterval)
  1616. }
  1617. sub.mu.Unlock()
  1618. // This call will return an error (which we don't care here)
  1619. // if nc is nil or fcReply is empty.
  1620. nc.Publish(fcReply, nil)
  1621. }
  1622. // ErrConsumerSequenceMismatch represents an error from a consumer
  1623. // that received a Heartbeat including sequence different to the
  1624. // one expected from the view of the client.
  1625. type ErrConsumerSequenceMismatch struct {
  1626. // StreamResumeSequence is the stream sequence from where the consumer
  1627. // should resume consuming from the stream.
  1628. StreamResumeSequence uint64
  1629. // ConsumerSequence is the sequence of the consumer that is behind.
  1630. ConsumerSequence uint64
  1631. // LastConsumerSequence is the sequence of the consumer when the heartbeat
  1632. // was received.
  1633. LastConsumerSequence uint64
  1634. }
  1635. func (ecs *ErrConsumerSequenceMismatch) Error() string {
  1636. return fmt.Sprintf("nats: sequence mismatch for consumer at sequence %d (%d sequences behind), should restart consumer from stream sequence %d",
  1637. ecs.ConsumerSequence,
  1638. ecs.LastConsumerSequence-ecs.ConsumerSequence,
  1639. ecs.StreamResumeSequence,
  1640. )
  1641. }
  1642. // isJSControlMessage will return true if this is an empty control status message
  1643. // and indicate what type of control message it is, say jsCtrlHB or jsCtrlFC
  1644. func isJSControlMessage(msg *Msg) (bool, int) {
  1645. if len(msg.Data) > 0 || msg.Header.Get(statusHdr) != controlMsg {
  1646. return false, 0
  1647. }
  1648. val := msg.Header.Get(descrHdr)
  1649. if strings.HasPrefix(val, "Idle") {
  1650. return true, jsCtrlHB
  1651. }
  1652. if strings.HasPrefix(val, "Flow") {
  1653. return true, jsCtrlFC
  1654. }
  1655. return true, 0
  1656. }
  1657. // Keeps track of the incoming message's reply subject so that the consumer's
  1658. // state (deliver sequence, etc..) can be checked against heartbeats.
  1659. // We will also bump the incoming data message sequence that is used in FC cases.
  1660. // Runs under the subscription lock
  1661. func (sub *Subscription) trackSequences(reply string) {
  1662. // For flow control, keep track of incoming message sequence.
  1663. sub.jsi.fciseq++
  1664. sub.jsi.cmeta = reply
  1665. }
  1666. // Check to make sure messages are arriving in order.
  1667. // Returns true if the sub had to be replaced. Will cause upper layers to return.
  1668. // The caller has verified that sub.jsi != nil and that this is not a control message.
  1669. // Lock should be held.
  1670. func (sub *Subscription) checkOrderedMsgs(m *Msg) bool {
  1671. // Ignore msgs with no reply like HBs and flow control, they are handled elsewhere.
  1672. if m.Reply == _EMPTY_ {
  1673. return false
  1674. }
  1675. // Normal message here.
  1676. tokens, err := parser.GetMetadataFields(m.Reply)
  1677. if err != nil {
  1678. return false
  1679. }
  1680. sseq, dseq := parser.ParseNum(tokens[ackStreamSeqTokenPos]), parser.ParseNum(tokens[ackConsumerSeqTokenPos])
  1681. jsi := sub.jsi
  1682. if dseq != jsi.dseq {
  1683. sub.resetOrderedConsumer(jsi.sseq + 1)
  1684. return true
  1685. }
  1686. // Update our tracking here.
  1687. jsi.dseq, jsi.sseq = dseq+1, sseq
  1688. return false
  1689. }
  1690. // Update and replace sid.
  1691. // Lock should be held on entry but will be unlocked to prevent lock inversion.
  1692. func (sub *Subscription) applyNewSID() (osid int64) {
  1693. nc := sub.conn
  1694. sub.mu.Unlock()
  1695. nc.subsMu.Lock()
  1696. osid = sub.sid
  1697. delete(nc.subs, osid)
  1698. // Place new one.
  1699. nc.ssid++
  1700. nsid := nc.ssid
  1701. nc.subs[nsid] = sub
  1702. nc.subsMu.Unlock()
  1703. sub.mu.Lock()
  1704. sub.sid = nsid
  1705. return osid
  1706. }
  1707. // We are here if we have detected a gap with an ordered consumer.
  1708. // We will create a new consumer and rewire the low level subscription.
  1709. // Lock should be held.
  1710. func (sub *Subscription) resetOrderedConsumer(sseq uint64) {
  1711. nc := sub.conn
  1712. if sub.jsi == nil || nc == nil || sub.closed {
  1713. return
  1714. }
  1715. var maxStr string
  1716. // If there was an AUTO_UNSUB done, we need to adjust the new value
  1717. // to send after the SUB for the new sid.
  1718. if sub.max > 0 {
  1719. if sub.jsi.fciseq < sub.max {
  1720. adjustedMax := sub.max - sub.jsi.fciseq
  1721. maxStr = strconv.Itoa(int(adjustedMax))
  1722. } else {
  1723. // We are already at the max, so we should just unsub the
  1724. // existing sub and be done
  1725. go func(sid int64) {
  1726. nc.mu.Lock()
  1727. nc.bw.appendString(fmt.Sprintf(unsubProto, sid, _EMPTY_))
  1728. nc.kickFlusher()
  1729. nc.mu.Unlock()
  1730. }(sub.sid)
  1731. return
  1732. }
  1733. }
  1734. // Quick unsubscribe. Since we know this is a simple push subscriber we do in place.
  1735. osid := sub.applyNewSID()
  1736. // Grab new inbox.
  1737. newDeliver := nc.NewInbox()
  1738. sub.Subject = newDeliver
  1739. // Snapshot the new sid under sub lock.
  1740. nsid := sub.sid
  1741. // We are still in the low level readLoop for the connection so we need
  1742. // to spin a go routine to try to create the new consumer.
  1743. go func() {
  1744. // Unsubscribe and subscribe with new inbox and sid.
  1745. // Remap a new low level sub into this sub since its client accessible.
  1746. // This is done here in this go routine to prevent lock inversion.
  1747. nc.mu.Lock()
  1748. nc.bw.appendString(fmt.Sprintf(unsubProto, osid, _EMPTY_))
  1749. nc.bw.appendString(fmt.Sprintf(subProto, newDeliver, _EMPTY_, nsid))
  1750. if maxStr != _EMPTY_ {
  1751. nc.bw.appendString(fmt.Sprintf(unsubProto, nsid, maxStr))
  1752. }
  1753. nc.kickFlusher()
  1754. nc.mu.Unlock()
  1755. pushErr := func(err error) {
  1756. nc.handleConsumerSequenceMismatch(sub, fmt.Errorf("%w: recreating ordered consumer", err))
  1757. nc.unsubscribe(sub, 0, true)
  1758. }
  1759. sub.mu.Lock()
  1760. jsi := sub.jsi
  1761. // Reset some items in jsi.
  1762. jsi.dseq = 1
  1763. jsi.cmeta = _EMPTY_
  1764. jsi.fcr, jsi.fcd = _EMPTY_, 0
  1765. jsi.deliver = newDeliver
  1766. // Reset consumer request for starting policy.
  1767. cfg := jsi.ccreq.Config
  1768. cfg.DeliverSubject = newDeliver
  1769. cfg.DeliverPolicy = DeliverByStartSequencePolicy
  1770. cfg.OptStartSeq = sseq
  1771. // In case the consumer was created with a start time, we need to clear it
  1772. // since we are now using a start sequence.
  1773. cfg.OptStartTime = nil
  1774. js := jsi.js
  1775. sub.mu.Unlock()
  1776. consName := nuid.Next()
  1777. cinfo, err := js.upsertConsumer(jsi.stream, consName, cfg)
  1778. if err != nil {
  1779. var apiErr *APIError
  1780. if errors.Is(err, ErrJetStreamNotEnabled) || errors.Is(err, ErrTimeout) {
  1781. // if creating consumer failed, retry
  1782. return
  1783. } else if errors.As(err, &apiErr) && apiErr.ErrorCode == JSErrCodeInsufficientResourcesErr {
  1784. // retry for insufficient resources, as it may mean that client is connected to a running
  1785. // server in cluster while the server hosting R1 JetStream resources is restarting
  1786. return
  1787. }
  1788. pushErr(err)
  1789. return
  1790. }
  1791. sub.mu.Lock()
  1792. jsi.consumer = cinfo.Name
  1793. sub.mu.Unlock()
  1794. }()
  1795. }
  1796. // For jetstream subscriptions, returns the number of delivered messages.
  1797. // For ChanSubscription, this value is computed based on the known number
  1798. // of messages added to the channel minus the current size of that channel.
  1799. // Lock held on entry
  1800. func (sub *Subscription) getJSDelivered() uint64 {
  1801. if sub.typ == ChanSubscription {
  1802. return sub.jsi.fciseq - uint64(len(sub.mch))
  1803. }
  1804. return sub.delivered
  1805. }
  1806. // checkForFlowControlResponse will check to see if we should send a flow control response
  1807. // based on the subscription current delivered index and the target.
  1808. // Runs under subscription lock
  1809. func (sub *Subscription) checkForFlowControlResponse() string {
  1810. // Caller has verified that there is a sub.jsi and fc
  1811. jsi := sub.jsi
  1812. jsi.active = true
  1813. if sub.getJSDelivered() >= jsi.fcd {
  1814. fcr := jsi.fcr
  1815. jsi.fcr, jsi.fcd = _EMPTY_, 0
  1816. return fcr
  1817. }
  1818. return _EMPTY_
  1819. }
  1820. // Record an inbound flow control message.
  1821. // Runs under subscription lock
  1822. func (sub *Subscription) scheduleFlowControlResponse(reply string) {
  1823. sub.jsi.fcr, sub.jsi.fcd = reply, sub.jsi.fciseq
  1824. }
  1825. // Checks for activity from our consumer.
  1826. // If we do not think we are active send an async error.
  1827. func (sub *Subscription) activityCheck() {
  1828. sub.mu.Lock()
  1829. jsi := sub.jsi
  1830. if jsi == nil || sub.closed {
  1831. sub.mu.Unlock()
  1832. return
  1833. }
  1834. active := jsi.active
  1835. jsi.hbc.Reset(jsi.hbi * hbcThresh)
  1836. jsi.active = false
  1837. nc := sub.conn
  1838. sub.mu.Unlock()
  1839. if !active {
  1840. if !jsi.ordered || nc.Status() != CONNECTED {
  1841. nc.mu.Lock()
  1842. if errCB := nc.Opts.AsyncErrorCB; errCB != nil {
  1843. nc.ach.push(func() { errCB(nc, sub, ErrConsumerNotActive) })
  1844. }
  1845. nc.mu.Unlock()
  1846. return
  1847. }
  1848. sub.mu.Lock()
  1849. sub.resetOrderedConsumer(jsi.sseq + 1)
  1850. sub.mu.Unlock()
  1851. }
  1852. }
  1853. // scheduleHeartbeatCheck sets up the timer check to make sure we are active
  1854. // or receiving idle heartbeats..
  1855. func (sub *Subscription) scheduleHeartbeatCheck() {
  1856. sub.mu.Lock()
  1857. defer sub.mu.Unlock()
  1858. jsi := sub.jsi
  1859. if jsi == nil {
  1860. return
  1861. }
  1862. if jsi.hbc == nil {
  1863. jsi.hbc = time.AfterFunc(jsi.hbi*hbcThresh, sub.activityCheck)
  1864. } else {
  1865. jsi.hbc.Reset(jsi.hbi * hbcThresh)
  1866. }
  1867. }
  1868. // handleConsumerSequenceMismatch will send an async error that can be used to restart a push based consumer.
  1869. func (nc *Conn) handleConsumerSequenceMismatch(sub *Subscription, err error) {
  1870. nc.mu.Lock()
  1871. errCB := nc.Opts.AsyncErrorCB
  1872. if errCB != nil {
  1873. nc.ach.push(func() { errCB(nc, sub, err) })
  1874. }
  1875. nc.mu.Unlock()
  1876. }
  1877. // checkForSequenceMismatch will make sure we have not missed any messages since last seen.
  1878. func (nc *Conn) checkForSequenceMismatch(msg *Msg, s *Subscription, jsi *jsSub) {
  1879. // Process heartbeat received, get latest control metadata if present.
  1880. s.mu.Lock()
  1881. ctrl, ordered := jsi.cmeta, jsi.ordered
  1882. jsi.active = true
  1883. s.mu.Unlock()
  1884. if ctrl == _EMPTY_ {
  1885. return
  1886. }
  1887. tokens, err := parser.GetMetadataFields(ctrl)
  1888. if err != nil {
  1889. return
  1890. }
  1891. // Consumer sequence.
  1892. var ldseq string
  1893. dseq := tokens[ackConsumerSeqTokenPos]
  1894. hdr := msg.Header[lastConsumerSeqHdr]
  1895. if len(hdr) == 1 {
  1896. ldseq = hdr[0]
  1897. }
  1898. // Detect consumer sequence mismatch and whether
  1899. // should restart the consumer.
  1900. if ldseq != dseq {
  1901. // Dispatch async error including details such as
  1902. // from where the consumer could be restarted.
  1903. sseq := parser.ParseNum(tokens[ackStreamSeqTokenPos])
  1904. if ordered {
  1905. s.mu.Lock()
  1906. s.resetOrderedConsumer(jsi.sseq + 1)
  1907. s.mu.Unlock()
  1908. } else {
  1909. ecs := &ErrConsumerSequenceMismatch{
  1910. StreamResumeSequence: uint64(sseq),
  1911. ConsumerSequence: parser.ParseNum(dseq),
  1912. LastConsumerSequence: parser.ParseNum(ldseq),
  1913. }
  1914. nc.handleConsumerSequenceMismatch(s, ecs)
  1915. }
  1916. }
  1917. }
  1918. type streamRequest struct {
  1919. Subject string `json:"subject,omitempty"`
  1920. }
  1921. type streamNamesResponse struct {
  1922. apiResponse
  1923. apiPaged
  1924. Streams []string `json:"streams"`
  1925. }
  1926. type subOpts struct {
  1927. // For attaching.
  1928. stream, consumer string
  1929. // For creating or updating.
  1930. cfg *ConsumerConfig
  1931. // For binding a subscription to a consumer without creating it.
  1932. bound bool
  1933. // For manual ack
  1934. mack bool
  1935. // For an ordered consumer.
  1936. ordered bool
  1937. ctx context.Context
  1938. // To disable calling ConsumerInfo
  1939. skipCInfo bool
  1940. }
  1941. // SkipConsumerLookup will omit lookipng up consumer when [Bind], [Durable]
  1942. // or [ConsumerName] are provided.
  1943. //
  1944. // NOTE: This setting may cause an existing consumer to be overwritten. Also,
  1945. // because consumer lookup is skipped, all consumer options like AckPolicy,
  1946. // DeliverSubject etc. need to be provided even if consumer already exists.
  1947. func SkipConsumerLookup() SubOpt {
  1948. return subOptFn(func(opts *subOpts) error {
  1949. opts.skipCInfo = true
  1950. return nil
  1951. })
  1952. }
  1953. // OrderedConsumer will create a FIFO direct/ephemeral consumer for in order delivery of messages.
  1954. // There are no redeliveries and no acks, and flow control and heartbeats will be added but
  1955. // will be taken care of without additional client code.
  1956. func OrderedConsumer() SubOpt {
  1957. return subOptFn(func(opts *subOpts) error {
  1958. opts.ordered = true
  1959. return nil
  1960. })
  1961. }
  1962. // ManualAck disables auto ack functionality for async subscriptions.
  1963. func ManualAck() SubOpt {
  1964. return subOptFn(func(opts *subOpts) error {
  1965. opts.mack = true
  1966. return nil
  1967. })
  1968. }
  1969. // Description will set the description for the created consumer.
  1970. func Description(description string) SubOpt {
  1971. return subOptFn(func(opts *subOpts) error {
  1972. opts.cfg.Description = description
  1973. return nil
  1974. })
  1975. }
  1976. // Durable defines the consumer name for JetStream durable subscribers.
  1977. // This function will return ErrInvalidConsumerName if the name contains
  1978. // any dot ".".
  1979. func Durable(consumer string) SubOpt {
  1980. return subOptFn(func(opts *subOpts) error {
  1981. if opts.cfg.Durable != _EMPTY_ {
  1982. return fmt.Errorf("nats: option Durable set more than once")
  1983. }
  1984. if opts.consumer != _EMPTY_ && opts.consumer != consumer {
  1985. return fmt.Errorf("nats: duplicate consumer names (%s and %s)", opts.consumer, consumer)
  1986. }
  1987. if err := checkConsumerName(consumer); err != nil {
  1988. return err
  1989. }
  1990. opts.cfg.Durable = consumer
  1991. return nil
  1992. })
  1993. }
  1994. // DeliverAll will configure a Consumer to receive all the
  1995. // messages from a Stream.
  1996. func DeliverAll() SubOpt {
  1997. return subOptFn(func(opts *subOpts) error {
  1998. opts.cfg.DeliverPolicy = DeliverAllPolicy
  1999. return nil
  2000. })
  2001. }
  2002. // DeliverLast configures a Consumer to receive messages
  2003. // starting with the latest one.
  2004. func DeliverLast() SubOpt {
  2005. return subOptFn(func(opts *subOpts) error {
  2006. opts.cfg.DeliverPolicy = DeliverLastPolicy
  2007. return nil
  2008. })
  2009. }
  2010. // DeliverLastPerSubject configures a Consumer to receive messages
  2011. // starting with the latest one for each filtered subject.
  2012. func DeliverLastPerSubject() SubOpt {
  2013. return subOptFn(func(opts *subOpts) error {
  2014. opts.cfg.DeliverPolicy = DeliverLastPerSubjectPolicy
  2015. return nil
  2016. })
  2017. }
  2018. // DeliverNew configures a Consumer to receive messages
  2019. // published after the subscription.
  2020. func DeliverNew() SubOpt {
  2021. return subOptFn(func(opts *subOpts) error {
  2022. opts.cfg.DeliverPolicy = DeliverNewPolicy
  2023. return nil
  2024. })
  2025. }
  2026. // StartSequence configures a Consumer to receive
  2027. // messages from a start sequence.
  2028. func StartSequence(seq uint64) SubOpt {
  2029. return subOptFn(func(opts *subOpts) error {
  2030. opts.cfg.DeliverPolicy = DeliverByStartSequencePolicy
  2031. opts.cfg.OptStartSeq = seq
  2032. return nil
  2033. })
  2034. }
  2035. // StartTime configures a Consumer to receive
  2036. // messages from a start time.
  2037. func StartTime(startTime time.Time) SubOpt {
  2038. return subOptFn(func(opts *subOpts) error {
  2039. opts.cfg.DeliverPolicy = DeliverByStartTimePolicy
  2040. opts.cfg.OptStartTime = &startTime
  2041. return nil
  2042. })
  2043. }
  2044. // AckNone requires no acks for delivered messages.
  2045. func AckNone() SubOpt {
  2046. return subOptFn(func(opts *subOpts) error {
  2047. opts.cfg.AckPolicy = AckNonePolicy
  2048. return nil
  2049. })
  2050. }
  2051. // AckAll when acking a sequence number, this implicitly acks all sequences
  2052. // below this one as well.
  2053. func AckAll() SubOpt {
  2054. return subOptFn(func(opts *subOpts) error {
  2055. opts.cfg.AckPolicy = AckAllPolicy
  2056. return nil
  2057. })
  2058. }
  2059. // AckExplicit requires ack or nack for all messages.
  2060. func AckExplicit() SubOpt {
  2061. return subOptFn(func(opts *subOpts) error {
  2062. opts.cfg.AckPolicy = AckExplicitPolicy
  2063. return nil
  2064. })
  2065. }
  2066. // MaxDeliver sets the number of redeliveries for a message.
  2067. func MaxDeliver(n int) SubOpt {
  2068. return subOptFn(func(opts *subOpts) error {
  2069. opts.cfg.MaxDeliver = n
  2070. return nil
  2071. })
  2072. }
  2073. // MaxAckPending sets the number of outstanding acks that are allowed before
  2074. // message delivery is halted.
  2075. func MaxAckPending(n int) SubOpt {
  2076. return subOptFn(func(opts *subOpts) error {
  2077. opts.cfg.MaxAckPending = n
  2078. return nil
  2079. })
  2080. }
  2081. // ReplayOriginal replays the messages at the original speed.
  2082. func ReplayOriginal() SubOpt {
  2083. return subOptFn(func(opts *subOpts) error {
  2084. opts.cfg.ReplayPolicy = ReplayOriginalPolicy
  2085. return nil
  2086. })
  2087. }
  2088. // ReplayInstant replays the messages as fast as possible.
  2089. func ReplayInstant() SubOpt {
  2090. return subOptFn(func(opts *subOpts) error {
  2091. opts.cfg.ReplayPolicy = ReplayInstantPolicy
  2092. return nil
  2093. })
  2094. }
  2095. // RateLimit is the Bits per sec rate limit applied to a push consumer.
  2096. func RateLimit(n uint64) SubOpt {
  2097. return subOptFn(func(opts *subOpts) error {
  2098. opts.cfg.RateLimit = n
  2099. return nil
  2100. })
  2101. }
  2102. // BackOff is an array of time durations that represent the time to delay based on delivery count.
  2103. func BackOff(backOff []time.Duration) SubOpt {
  2104. return subOptFn(func(opts *subOpts) error {
  2105. opts.cfg.BackOff = backOff
  2106. return nil
  2107. })
  2108. }
  2109. // BindStream binds a consumer to a stream explicitly based on a name.
  2110. // When a stream name is not specified, the library uses the subscribe
  2111. // subject as a way to find the stream name. It is done by making a request
  2112. // to the server to get list of stream names that have a filter for this
  2113. // subject. If the returned list contains a single stream, then this
  2114. // stream name will be used, otherwise the `ErrNoMatchingStream` is returned.
  2115. // To avoid the stream lookup, provide the stream name with this function.
  2116. // See also `Bind()`.
  2117. func BindStream(stream string) SubOpt {
  2118. return subOptFn(func(opts *subOpts) error {
  2119. if opts.stream != _EMPTY_ && opts.stream != stream {
  2120. return fmt.Errorf("nats: duplicate stream name (%s and %s)", opts.stream, stream)
  2121. }
  2122. opts.stream = stream
  2123. return nil
  2124. })
  2125. }
  2126. // Bind binds a subscription to an existing consumer from a stream without attempting to create.
  2127. // The first argument is the stream name and the second argument will be the consumer name.
  2128. func Bind(stream, consumer string) SubOpt {
  2129. return subOptFn(func(opts *subOpts) error {
  2130. if stream == _EMPTY_ {
  2131. return ErrStreamNameRequired
  2132. }
  2133. if consumer == _EMPTY_ {
  2134. return ErrConsumerNameRequired
  2135. }
  2136. // In case of pull subscribers, the durable name is a required parameter
  2137. // so check that they are not different.
  2138. if opts.cfg.Durable != _EMPTY_ && opts.cfg.Durable != consumer {
  2139. return fmt.Errorf("nats: duplicate consumer names (%s and %s)", opts.cfg.Durable, consumer)
  2140. }
  2141. if opts.stream != _EMPTY_ && opts.stream != stream {
  2142. return fmt.Errorf("nats: duplicate stream name (%s and %s)", opts.stream, stream)
  2143. }
  2144. opts.stream = stream
  2145. opts.consumer = consumer
  2146. opts.bound = true
  2147. return nil
  2148. })
  2149. }
  2150. // EnableFlowControl enables flow control for a push based consumer.
  2151. func EnableFlowControl() SubOpt {
  2152. return subOptFn(func(opts *subOpts) error {
  2153. opts.cfg.FlowControl = true
  2154. return nil
  2155. })
  2156. }
  2157. // IdleHeartbeat enables push based consumers to have idle heartbeats delivered.
  2158. func IdleHeartbeat(duration time.Duration) SubOpt {
  2159. return subOptFn(func(opts *subOpts) error {
  2160. opts.cfg.Heartbeat = duration
  2161. return nil
  2162. })
  2163. }
  2164. // DeliverSubject specifies the JetStream consumer deliver subject.
  2165. //
  2166. // This option is used only in situations where the consumer does not exist
  2167. // and a creation request is sent to the server. If not provided, an inbox
  2168. // will be selected.
  2169. // If a consumer exists, then the NATS subscription will be created on
  2170. // the JetStream consumer's DeliverSubject, not necessarily this subject.
  2171. func DeliverSubject(subject string) SubOpt {
  2172. return subOptFn(func(opts *subOpts) error {
  2173. opts.cfg.DeliverSubject = subject
  2174. return nil
  2175. })
  2176. }
  2177. // HeadersOnly() will instruct the consumer to only deliver headers and no payloads.
  2178. func HeadersOnly() SubOpt {
  2179. return subOptFn(func(opts *subOpts) error {
  2180. opts.cfg.HeadersOnly = true
  2181. return nil
  2182. })
  2183. }
  2184. // MaxRequestBatch sets the maximum pull consumer batch size that a Fetch()
  2185. // can request.
  2186. func MaxRequestBatch(max int) SubOpt {
  2187. return subOptFn(func(opts *subOpts) error {
  2188. opts.cfg.MaxRequestBatch = max
  2189. return nil
  2190. })
  2191. }
  2192. // MaxRequestExpires sets the maximum pull consumer request expiration that a
  2193. // Fetch() can request (using the Fetch's timeout value).
  2194. func MaxRequestExpires(max time.Duration) SubOpt {
  2195. return subOptFn(func(opts *subOpts) error {
  2196. opts.cfg.MaxRequestExpires = max
  2197. return nil
  2198. })
  2199. }
  2200. // MaxRequesMaxBytes sets the maximum pull consumer request bytes that a
  2201. // Fetch() can receive.
  2202. func MaxRequestMaxBytes(bytes int) SubOpt {
  2203. return subOptFn(func(opts *subOpts) error {
  2204. opts.cfg.MaxRequestMaxBytes = bytes
  2205. return nil
  2206. })
  2207. }
  2208. // InactiveThreshold indicates how long the server should keep a consumer
  2209. // after detecting a lack of activity. In NATS Server 2.8.4 and earlier, this
  2210. // option only applies to ephemeral consumers. In NATS Server 2.9.0 and later,
  2211. // this option applies to both ephemeral and durable consumers, allowing durable
  2212. // consumers to also be deleted automatically after the inactivity threshold has
  2213. // passed.
  2214. func InactiveThreshold(threshold time.Duration) SubOpt {
  2215. return subOptFn(func(opts *subOpts) error {
  2216. if threshold < 0 {
  2217. return fmt.Errorf("invalid InactiveThreshold value (%v), needs to be greater or equal to 0", threshold)
  2218. }
  2219. opts.cfg.InactiveThreshold = threshold
  2220. return nil
  2221. })
  2222. }
  2223. // ConsumerReplicas sets the number of replica count for a consumer.
  2224. func ConsumerReplicas(replicas int) SubOpt {
  2225. return subOptFn(func(opts *subOpts) error {
  2226. if replicas < 1 {
  2227. return fmt.Errorf("invalid ConsumerReplicas value (%v), needs to be greater than 0", replicas)
  2228. }
  2229. opts.cfg.Replicas = replicas
  2230. return nil
  2231. })
  2232. }
  2233. // ConsumerMemoryStorage sets the memory storage to true for a consumer.
  2234. func ConsumerMemoryStorage() SubOpt {
  2235. return subOptFn(func(opts *subOpts) error {
  2236. opts.cfg.MemoryStorage = true
  2237. return nil
  2238. })
  2239. }
  2240. // ConsumerName sets the name for a consumer.
  2241. func ConsumerName(name string) SubOpt {
  2242. return subOptFn(func(opts *subOpts) error {
  2243. opts.cfg.Name = name
  2244. return nil
  2245. })
  2246. }
  2247. func (sub *Subscription) ConsumerInfo() (*ConsumerInfo, error) {
  2248. sub.mu.Lock()
  2249. // TODO(dlc) - Better way to mark especially if we attach.
  2250. if sub.jsi == nil || sub.jsi.consumer == _EMPTY_ {
  2251. sub.mu.Unlock()
  2252. return nil, ErrTypeSubscription
  2253. }
  2254. // Consumer info lookup should fail if in direct mode.
  2255. js := sub.jsi.js
  2256. stream, consumer := sub.jsi.stream, sub.jsi.consumer
  2257. sub.mu.Unlock()
  2258. return js.getConsumerInfo(stream, consumer)
  2259. }
  2260. type pullOpts struct {
  2261. maxBytes int
  2262. ttl time.Duration
  2263. ctx context.Context
  2264. }
  2265. // PullOpt are the options that can be passed when pulling a batch of messages.
  2266. type PullOpt interface {
  2267. configurePull(opts *pullOpts) error
  2268. }
  2269. // PullMaxWaiting defines the max inflight pull requests.
  2270. func PullMaxWaiting(n int) SubOpt {
  2271. return subOptFn(func(opts *subOpts) error {
  2272. opts.cfg.MaxWaiting = n
  2273. return nil
  2274. })
  2275. }
  2276. // PullMaxBytes defines the max bytes allowed for a fetch request.
  2277. type PullMaxBytes int
  2278. func (n PullMaxBytes) configurePull(opts *pullOpts) error {
  2279. opts.maxBytes = int(n)
  2280. return nil
  2281. }
  2282. var (
  2283. // errNoMessages is an error that a Fetch request using no_wait can receive to signal
  2284. // that there are no more messages available.
  2285. errNoMessages = errors.New("nats: no messages")
  2286. // errRequestsPending is an error that represents a sub.Fetch requests that was using
  2287. // no_wait and expires time got discarded by the server.
  2288. errRequestsPending = errors.New("nats: requests pending")
  2289. )
  2290. // Returns if the given message is a user message or not, and if
  2291. // `checkSts` is true, returns appropriate error based on the
  2292. // content of the status (404, etc..)
  2293. func checkMsg(msg *Msg, checkSts, isNoWait bool) (usrMsg bool, err error) {
  2294. // Assume user message
  2295. usrMsg = true
  2296. // If payload or no header, consider this a user message
  2297. if len(msg.Data) > 0 || len(msg.Header) == 0 {
  2298. return
  2299. }
  2300. // Look for status header
  2301. val := msg.Header.Get(statusHdr)
  2302. // If not present, then this is considered a user message
  2303. if val == _EMPTY_ {
  2304. return
  2305. }
  2306. // At this point, this is not a user message since there is
  2307. // no payload and a "Status" header.
  2308. usrMsg = false
  2309. // If we don't care about status, we are done.
  2310. if !checkSts {
  2311. return
  2312. }
  2313. switch val {
  2314. case noResponders:
  2315. err = ErrNoResponders
  2316. case noMessagesSts:
  2317. // 404 indicates that there are no messages.
  2318. err = errNoMessages
  2319. case reqTimeoutSts:
  2320. // In case of a fetch request with no wait request and expires time,
  2321. // need to skip 408 errors and retry.
  2322. if isNoWait {
  2323. err = errRequestsPending
  2324. } else {
  2325. // Older servers may send a 408 when a request in the server was expired
  2326. // and interest is still found, which will be the case for our
  2327. // implementation. Regardless, ignore 408 errors until receiving at least
  2328. // one message when making requests without no_wait.
  2329. err = ErrTimeout
  2330. }
  2331. case jetStream409Sts:
  2332. if strings.Contains(strings.ToLower(msg.Header.Get(descrHdr)), "consumer deleted") {
  2333. err = ErrConsumerDeleted
  2334. break
  2335. }
  2336. if strings.Contains(strings.ToLower(msg.Header.Get(descrHdr)), "leadership change") {
  2337. err = ErrConsumerLeadershipChanged
  2338. break
  2339. }
  2340. fallthrough
  2341. default:
  2342. err = fmt.Errorf("nats: %s", msg.Header.Get(descrHdr))
  2343. }
  2344. return
  2345. }
  2346. // Fetch pulls a batch of messages from a stream for a pull consumer.
  2347. func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
  2348. if sub == nil {
  2349. return nil, ErrBadSubscription
  2350. }
  2351. if batch < 1 {
  2352. return nil, ErrInvalidArg
  2353. }
  2354. var o pullOpts
  2355. for _, opt := range opts {
  2356. if err := opt.configurePull(&o); err != nil {
  2357. return nil, err
  2358. }
  2359. }
  2360. if o.ctx != nil && o.ttl != 0 {
  2361. return nil, ErrContextAndTimeout
  2362. }
  2363. sub.mu.Lock()
  2364. jsi := sub.jsi
  2365. // Reject if this is not a pull subscription. Note that sub.typ is SyncSubscription,
  2366. // so check for jsi.pull boolean instead.
  2367. if jsi == nil || !jsi.pull {
  2368. sub.mu.Unlock()
  2369. return nil, ErrTypeSubscription
  2370. }
  2371. nc := sub.conn
  2372. nms := sub.jsi.nms
  2373. rply, _ := newFetchInbox(jsi.deliver)
  2374. js := sub.jsi.js
  2375. pmc := len(sub.mch) > 0
  2376. // All fetch requests have an expiration, in case of no explicit expiration
  2377. // then the default timeout of the JetStream context is used.
  2378. ttl := o.ttl
  2379. if ttl == 0 {
  2380. ttl = js.opts.wait
  2381. }
  2382. sub.mu.Unlock()
  2383. // Use the given context or setup a default one for the span
  2384. // of the pull batch request.
  2385. var (
  2386. ctx = o.ctx
  2387. err error
  2388. cancel context.CancelFunc
  2389. )
  2390. if ctx == nil {
  2391. ctx, cancel = context.WithTimeout(context.Background(), ttl)
  2392. defer cancel()
  2393. } else if _, hasDeadline := ctx.Deadline(); !hasDeadline {
  2394. // Prevent from passing the background context which will just block
  2395. // and cannot be canceled either.
  2396. if octx, ok := ctx.(ContextOpt); ok && octx.Context == context.Background() {
  2397. return nil, ErrNoDeadlineContext
  2398. }
  2399. // If the context did not have a deadline, then create a new child context
  2400. // that will use the default timeout from the JS context.
  2401. ctx, cancel = context.WithTimeout(ctx, ttl)
  2402. defer cancel()
  2403. }
  2404. // Check if context not done already before making the request.
  2405. select {
  2406. case <-ctx.Done():
  2407. if o.ctx != nil { // Timeout or Cancel triggered by context object option
  2408. err = ctx.Err()
  2409. } else { // Timeout triggered by timeout option
  2410. err = ErrTimeout
  2411. }
  2412. default:
  2413. }
  2414. if err != nil {
  2415. return nil, err
  2416. }
  2417. var (
  2418. msgs = make([]*Msg, 0, batch)
  2419. msg *Msg
  2420. )
  2421. for pmc && len(msgs) < batch {
  2422. // Check next msg with booleans that say that this is an internal call
  2423. // for a pull subscribe (so don't reject it) and don't wait if there
  2424. // are no messages.
  2425. msg, err = sub.nextMsgWithContext(ctx, true, false)
  2426. if err != nil {
  2427. if err == errNoMessages {
  2428. err = nil
  2429. }
  2430. break
  2431. }
  2432. // Check msg but just to determine if this is a user message
  2433. // or status message, however, we don't care about values of status
  2434. // messages at this point in the Fetch() call, so checkMsg can't
  2435. // return an error.
  2436. if usrMsg, _ := checkMsg(msg, false, false); usrMsg {
  2437. msgs = append(msgs, msg)
  2438. }
  2439. }
  2440. if err == nil && len(msgs) < batch {
  2441. // For batch real size of 1, it does not make sense to set no_wait in
  2442. // the request.
  2443. noWait := batch-len(msgs) > 1
  2444. var nr nextRequest
  2445. sendReq := func() error {
  2446. // The current deadline for the context will be used
  2447. // to set the expires TTL for a fetch request.
  2448. deadline, _ := ctx.Deadline()
  2449. ttl = time.Until(deadline)
  2450. // Check if context has already been canceled or expired.
  2451. select {
  2452. case <-ctx.Done():
  2453. return ctx.Err()
  2454. default:
  2455. }
  2456. // Make our request expiration a bit shorter than the current timeout.
  2457. expires := ttl
  2458. if ttl >= 20*time.Millisecond {
  2459. expires = ttl - 10*time.Millisecond
  2460. }
  2461. nr.Batch = batch - len(msgs)
  2462. nr.Expires = expires
  2463. nr.NoWait = noWait
  2464. nr.MaxBytes = o.maxBytes
  2465. req, _ := json.Marshal(nr)
  2466. return nc.PublishRequest(nms, rply, req)
  2467. }
  2468. err = sendReq()
  2469. for err == nil && len(msgs) < batch {
  2470. // Ask for next message and wait if there are no messages
  2471. msg, err = sub.nextMsgWithContext(ctx, true, true)
  2472. if err == nil {
  2473. var usrMsg bool
  2474. usrMsg, err = checkMsg(msg, true, noWait)
  2475. if err == nil && usrMsg {
  2476. msgs = append(msgs, msg)
  2477. } else if noWait && (err == errNoMessages || err == errRequestsPending) && len(msgs) == 0 {
  2478. // If we have a 404/408 for our "no_wait" request and have
  2479. // not collected any message, then resend request to
  2480. // wait this time.
  2481. noWait = false
  2482. err = sendReq()
  2483. } else if err == ErrTimeout && len(msgs) == 0 {
  2484. // If we get a 408, we will bail if we already collected some
  2485. // messages, otherwise ignore and go back calling nextMsg.
  2486. err = nil
  2487. }
  2488. }
  2489. }
  2490. }
  2491. // If there is at least a message added to msgs, then need to return OK and no error
  2492. if err != nil && len(msgs) == 0 {
  2493. return nil, o.checkCtxErr(err)
  2494. }
  2495. return msgs, nil
  2496. }
  2497. // newFetchInbox returns subject used as reply subject when sending pull requests
  2498. // as well as request ID. For non-wildcard subject, request ID is empty and
  2499. // passed subject is not transformed
  2500. func newFetchInbox(subj string) (string, string) {
  2501. if !strings.HasSuffix(subj, ".*") {
  2502. return subj, ""
  2503. }
  2504. reqID := nuid.Next()
  2505. var sb strings.Builder
  2506. sb.WriteString(subj[:len(subj)-1])
  2507. sb.WriteString(reqID)
  2508. return sb.String(), reqID
  2509. }
  2510. func subjectMatchesReqID(subject, reqID string) bool {
  2511. subjectParts := strings.Split(subject, ".")
  2512. if len(subjectParts) < 2 {
  2513. return false
  2514. }
  2515. return subjectParts[len(subjectParts)-1] == reqID
  2516. }
  2517. // MessageBatch provides methods to retrieve messages consumed using [Subscribe.FetchBatch].
  2518. type MessageBatch interface {
  2519. // Messages returns a channel on which messages will be published.
  2520. Messages() <-chan *Msg
  2521. // Error returns an error encountered when fetching messages.
  2522. Error() error
  2523. // Done signals end of execution.
  2524. Done() <-chan struct{}
  2525. }
  2526. type messageBatch struct {
  2527. msgs chan *Msg
  2528. err error
  2529. done chan struct{}
  2530. }
  2531. func (mb *messageBatch) Messages() <-chan *Msg {
  2532. return mb.msgs
  2533. }
  2534. func (mb *messageBatch) Error() error {
  2535. return mb.err
  2536. }
  2537. func (mb *messageBatch) Done() <-chan struct{} {
  2538. return mb.done
  2539. }
  2540. // FetchBatch pulls a batch of messages from a stream for a pull consumer.
  2541. // Unlike [Subscription.Fetch], it is non blocking and returns [MessageBatch],
  2542. // allowing to retrieve incoming messages from a channel.
  2543. // The returned channel is always closed after all messages for a batch have been
  2544. // delivered by the server - it is safe to iterate over it using range.
  2545. //
  2546. // To avoid using default JetStream timeout as fetch expiry time, use [nats.MaxWait]
  2547. // or [nats.Context] (with deadline set).
  2548. //
  2549. // This method will not return error in case of pull request expiry (even if there are no messages).
  2550. // Any other error encountered when receiving messages will cause FetchBatch to stop receiving new messages.
  2551. func (sub *Subscription) FetchBatch(batch int, opts ...PullOpt) (MessageBatch, error) {
  2552. if sub == nil {
  2553. return nil, ErrBadSubscription
  2554. }
  2555. if batch < 1 {
  2556. return nil, ErrInvalidArg
  2557. }
  2558. var o pullOpts
  2559. for _, opt := range opts {
  2560. if err := opt.configurePull(&o); err != nil {
  2561. return nil, err
  2562. }
  2563. }
  2564. if o.ctx != nil && o.ttl != 0 {
  2565. return nil, ErrContextAndTimeout
  2566. }
  2567. sub.mu.Lock()
  2568. jsi := sub.jsi
  2569. // Reject if this is not a pull subscription. Note that sub.typ is SyncSubscription,
  2570. // so check for jsi.pull boolean instead.
  2571. if jsi == nil || !jsi.pull {
  2572. sub.mu.Unlock()
  2573. return nil, ErrTypeSubscription
  2574. }
  2575. nc := sub.conn
  2576. nms := sub.jsi.nms
  2577. rply, reqID := newFetchInbox(sub.jsi.deliver)
  2578. js := sub.jsi.js
  2579. pmc := len(sub.mch) > 0
  2580. // All fetch requests have an expiration, in case of no explicit expiration
  2581. // then the default timeout of the JetStream context is used.
  2582. ttl := o.ttl
  2583. if ttl == 0 {
  2584. ttl = js.opts.wait
  2585. }
  2586. sub.mu.Unlock()
  2587. // Use the given context or setup a default one for the span
  2588. // of the pull batch request.
  2589. var (
  2590. ctx = o.ctx
  2591. cancel context.CancelFunc
  2592. cancelContext = true
  2593. )
  2594. if ctx == nil {
  2595. ctx, cancel = context.WithTimeout(context.Background(), ttl)
  2596. } else if _, hasDeadline := ctx.Deadline(); !hasDeadline {
  2597. // Prevent from passing the background context which will just block
  2598. // and cannot be canceled either.
  2599. if octx, ok := ctx.(ContextOpt); ok && octx.Context == context.Background() {
  2600. return nil, ErrNoDeadlineContext
  2601. }
  2602. // If the context did not have a deadline, then create a new child context
  2603. // that will use the default timeout from the JS context.
  2604. ctx, cancel = context.WithTimeout(ctx, ttl)
  2605. }
  2606. defer func() {
  2607. // only cancel the context here if we are sure the fetching goroutine has not been started yet
  2608. if cancel != nil && cancelContext {
  2609. cancel()
  2610. }
  2611. }()
  2612. // Check if context not done already before making the request.
  2613. select {
  2614. case <-ctx.Done():
  2615. if o.ctx != nil { // Timeout or Cancel triggered by context object option
  2616. return nil, ctx.Err()
  2617. } else { // Timeout triggered by timeout option
  2618. return nil, ErrTimeout
  2619. }
  2620. default:
  2621. }
  2622. result := &messageBatch{
  2623. msgs: make(chan *Msg, batch),
  2624. done: make(chan struct{}, 1),
  2625. }
  2626. var msg *Msg
  2627. for pmc && len(result.msgs) < batch {
  2628. // Check next msg with booleans that say that this is an internal call
  2629. // for a pull subscribe (so don't reject it) and don't wait if there
  2630. // are no messages.
  2631. msg, err := sub.nextMsgWithContext(ctx, true, false)
  2632. if err != nil {
  2633. if err == errNoMessages {
  2634. err = nil
  2635. }
  2636. result.err = err
  2637. break
  2638. }
  2639. // Check msg but just to determine if this is a user message
  2640. // or status message, however, we don't care about values of status
  2641. // messages at this point in the Fetch() call, so checkMsg can't
  2642. // return an error.
  2643. if usrMsg, _ := checkMsg(msg, false, false); usrMsg {
  2644. result.msgs <- msg
  2645. }
  2646. }
  2647. if len(result.msgs) == batch || result.err != nil {
  2648. close(result.msgs)
  2649. result.done <- struct{}{}
  2650. return result, nil
  2651. }
  2652. deadline, _ := ctx.Deadline()
  2653. ttl = time.Until(deadline)
  2654. // Make our request expiration a bit shorter than the current timeout.
  2655. expires := ttl
  2656. if ttl >= 20*time.Millisecond {
  2657. expires = ttl - 10*time.Millisecond
  2658. }
  2659. requestBatch := batch - len(result.msgs)
  2660. req := nextRequest{
  2661. Expires: expires,
  2662. Batch: requestBatch,
  2663. MaxBytes: o.maxBytes,
  2664. }
  2665. reqJSON, err := json.Marshal(req)
  2666. if err != nil {
  2667. close(result.msgs)
  2668. result.done <- struct{}{}
  2669. result.err = err
  2670. return result, nil
  2671. }
  2672. if err := nc.PublishRequest(nms, rply, reqJSON); err != nil {
  2673. if len(result.msgs) == 0 {
  2674. return nil, err
  2675. }
  2676. close(result.msgs)
  2677. result.done <- struct{}{}
  2678. result.err = err
  2679. return result, nil
  2680. }
  2681. cancelContext = false
  2682. go func() {
  2683. if cancel != nil {
  2684. defer cancel()
  2685. }
  2686. var requestMsgs int
  2687. for requestMsgs < requestBatch {
  2688. // Ask for next message and wait if there are no messages
  2689. msg, err = sub.nextMsgWithContext(ctx, true, true)
  2690. if err != nil {
  2691. break
  2692. }
  2693. var usrMsg bool
  2694. usrMsg, err = checkMsg(msg, true, false)
  2695. if err != nil {
  2696. if err == ErrTimeout {
  2697. if reqID != "" && !subjectMatchesReqID(msg.Subject, reqID) {
  2698. // ignore timeout message from server if it comes from a different pull request
  2699. continue
  2700. }
  2701. err = nil
  2702. }
  2703. break
  2704. }
  2705. if usrMsg {
  2706. result.msgs <- msg
  2707. requestMsgs++
  2708. }
  2709. }
  2710. if err != nil {
  2711. result.err = o.checkCtxErr(err)
  2712. }
  2713. close(result.msgs)
  2714. result.done <- struct{}{}
  2715. }()
  2716. return result, nil
  2717. }
  2718. // checkCtxErr is used to determine whether ErrTimeout should be returned in case of context timeout
  2719. func (o *pullOpts) checkCtxErr(err error) error {
  2720. if o.ctx == nil && err == context.DeadlineExceeded {
  2721. return ErrTimeout
  2722. }
  2723. return err
  2724. }
  2725. func (js *js) getConsumerInfo(stream, consumer string) (*ConsumerInfo, error) {
  2726. ctx, cancel := context.WithTimeout(context.Background(), js.opts.wait)
  2727. defer cancel()
  2728. return js.getConsumerInfoContext(ctx, stream, consumer)
  2729. }
  2730. func (js *js) getConsumerInfoContext(ctx context.Context, stream, consumer string) (*ConsumerInfo, error) {
  2731. ccInfoSubj := fmt.Sprintf(apiConsumerInfoT, stream, consumer)
  2732. resp, err := js.apiRequestWithContext(ctx, js.apiSubj(ccInfoSubj), nil)
  2733. if err != nil {
  2734. if err == ErrNoResponders {
  2735. err = ErrJetStreamNotEnabled
  2736. }
  2737. return nil, err
  2738. }
  2739. var info consumerResponse
  2740. if err := json.Unmarshal(resp.Data, &info); err != nil {
  2741. return nil, err
  2742. }
  2743. if info.Error != nil {
  2744. if errors.Is(info.Error, ErrConsumerNotFound) {
  2745. return nil, ErrConsumerNotFound
  2746. }
  2747. if errors.Is(info.Error, ErrStreamNotFound) {
  2748. return nil, ErrStreamNotFound
  2749. }
  2750. return nil, info.Error
  2751. }
  2752. return info.ConsumerInfo, nil
  2753. }
  2754. // a RequestWithContext with tracing via TraceCB
  2755. func (js *js) apiRequestWithContext(ctx context.Context, subj string, data []byte) (*Msg, error) {
  2756. if js.opts.shouldTrace {
  2757. ctrace := js.opts.ctrace
  2758. if ctrace.RequestSent != nil {
  2759. ctrace.RequestSent(subj, data)
  2760. }
  2761. }
  2762. resp, err := js.nc.RequestWithContext(ctx, subj, data)
  2763. if err != nil {
  2764. return nil, err
  2765. }
  2766. if js.opts.shouldTrace {
  2767. ctrace := js.opts.ctrace
  2768. if ctrace.RequestSent != nil {
  2769. ctrace.ResponseReceived(subj, resp.Data, resp.Header)
  2770. }
  2771. }
  2772. return resp, nil
  2773. }
  2774. func (m *Msg) checkReply() error {
  2775. if m == nil || m.Sub == nil {
  2776. return ErrMsgNotBound
  2777. }
  2778. if m.Reply == _EMPTY_ {
  2779. return ErrMsgNoReply
  2780. }
  2781. return nil
  2782. }
  2783. // ackReply handles all acks. Will do the right thing for pull and sync mode.
  2784. // It ensures that an ack is only sent a single time, regardless of
  2785. // how many times it is being called to avoid duplicated acks.
  2786. func (m *Msg) ackReply(ackType []byte, sync bool, opts ...AckOpt) error {
  2787. var o ackOpts
  2788. for _, opt := range opts {
  2789. if err := opt.configureAck(&o); err != nil {
  2790. return err
  2791. }
  2792. }
  2793. if err := m.checkReply(); err != nil {
  2794. return err
  2795. }
  2796. var ackNone bool
  2797. var js *js
  2798. sub := m.Sub
  2799. sub.mu.Lock()
  2800. nc := sub.conn
  2801. if jsi := sub.jsi; jsi != nil {
  2802. js = jsi.js
  2803. ackNone = jsi.ackNone
  2804. }
  2805. sub.mu.Unlock()
  2806. // Skip if already acked.
  2807. if atomic.LoadUint32(&m.ackd) == 1 {
  2808. return ErrMsgAlreadyAckd
  2809. }
  2810. if ackNone {
  2811. return ErrCantAckIfConsumerAckNone
  2812. }
  2813. usesCtx := o.ctx != nil
  2814. usesWait := o.ttl > 0
  2815. // Only allow either AckWait or Context option to set the timeout.
  2816. if usesWait && usesCtx {
  2817. return ErrContextAndTimeout
  2818. }
  2819. sync = sync || usesCtx || usesWait
  2820. ctx := o.ctx
  2821. wait := defaultRequestWait
  2822. if usesWait {
  2823. wait = o.ttl
  2824. } else if js != nil {
  2825. wait = js.opts.wait
  2826. }
  2827. var body []byte
  2828. var err error
  2829. // This will be > 0 only when called from NakWithDelay()
  2830. if o.nakDelay > 0 {
  2831. body = []byte(fmt.Sprintf("%s {\"delay\": %d}", ackType, o.nakDelay.Nanoseconds()))
  2832. } else {
  2833. body = ackType
  2834. }
  2835. if sync {
  2836. if usesCtx {
  2837. _, err = nc.RequestWithContext(ctx, m.Reply, body)
  2838. } else {
  2839. _, err = nc.Request(m.Reply, body, wait)
  2840. }
  2841. } else {
  2842. err = nc.Publish(m.Reply, body)
  2843. }
  2844. // Mark that the message has been acked unless it is ackProgress
  2845. // which can be sent many times.
  2846. if err == nil && !bytes.Equal(ackType, ackProgress) {
  2847. atomic.StoreUint32(&m.ackd, 1)
  2848. }
  2849. return err
  2850. }
  2851. // Ack acknowledges a message. This tells the server that the message was
  2852. // successfully processed and it can move on to the next message.
  2853. func (m *Msg) Ack(opts ...AckOpt) error {
  2854. return m.ackReply(ackAck, false, opts...)
  2855. }
  2856. // AckSync is the synchronous version of Ack. This indicates successful message
  2857. // processing.
  2858. func (m *Msg) AckSync(opts ...AckOpt) error {
  2859. return m.ackReply(ackAck, true, opts...)
  2860. }
  2861. // Nak negatively acknowledges a message. This tells the server to redeliver
  2862. // the message. You can configure the number of redeliveries by passing
  2863. // nats.MaxDeliver when you Subscribe. The default is infinite redeliveries.
  2864. func (m *Msg) Nak(opts ...AckOpt) error {
  2865. return m.ackReply(ackNak, false, opts...)
  2866. }
  2867. // Nak negatively acknowledges a message. This tells the server to redeliver
  2868. // the message after the give `delay` duration. You can configure the number
  2869. // of redeliveries by passing nats.MaxDeliver when you Subscribe.
  2870. // The default is infinite redeliveries.
  2871. func (m *Msg) NakWithDelay(delay time.Duration, opts ...AckOpt) error {
  2872. if delay > 0 {
  2873. opts = append(opts, nakDelay(delay))
  2874. }
  2875. return m.ackReply(ackNak, false, opts...)
  2876. }
  2877. // Term tells the server to not redeliver this message, regardless of the value
  2878. // of nats.MaxDeliver.
  2879. func (m *Msg) Term(opts ...AckOpt) error {
  2880. return m.ackReply(ackTerm, false, opts...)
  2881. }
  2882. // InProgress tells the server that this message is being worked on. It resets
  2883. // the redelivery timer on the server.
  2884. func (m *Msg) InProgress(opts ...AckOpt) error {
  2885. return m.ackReply(ackProgress, false, opts...)
  2886. }
  2887. // MsgMetadata is the JetStream metadata associated with received messages.
  2888. type MsgMetadata struct {
  2889. Sequence SequencePair
  2890. NumDelivered uint64
  2891. NumPending uint64
  2892. Timestamp time.Time
  2893. Stream string
  2894. Consumer string
  2895. Domain string
  2896. }
  2897. const (
  2898. ackDomainTokenPos = 2
  2899. ackAccHashTokenPos = 3
  2900. ackStreamTokenPos = 4
  2901. ackConsumerTokenPos = 5
  2902. ackNumDeliveredTokenPos = 6
  2903. ackStreamSeqTokenPos = 7
  2904. ackConsumerSeqTokenPos = 8
  2905. ackTimestampSeqTokenPos = 9
  2906. ackNumPendingTokenPos = 10
  2907. )
  2908. // Metadata retrieves the metadata from a JetStream message. This method will
  2909. // return an error for non-JetStream Msgs.
  2910. func (m *Msg) Metadata() (*MsgMetadata, error) {
  2911. if err := m.checkReply(); err != nil {
  2912. return nil, err
  2913. }
  2914. tokens, err := parser.GetMetadataFields(m.Reply)
  2915. if err != nil {
  2916. return nil, err
  2917. }
  2918. meta := &MsgMetadata{
  2919. Domain: tokens[ackDomainTokenPos],
  2920. NumDelivered: parser.ParseNum(tokens[ackNumDeliveredTokenPos]),
  2921. NumPending: parser.ParseNum(tokens[ackNumPendingTokenPos]),
  2922. Timestamp: time.Unix(0, int64(parser.ParseNum(tokens[ackTimestampSeqTokenPos]))),
  2923. Stream: tokens[ackStreamTokenPos],
  2924. Consumer: tokens[ackConsumerTokenPos],
  2925. }
  2926. meta.Sequence.Stream = parser.ParseNum(tokens[ackStreamSeqTokenPos])
  2927. meta.Sequence.Consumer = parser.ParseNum(tokens[ackConsumerSeqTokenPos])
  2928. return meta, nil
  2929. }
  2930. // AckPolicy determines how the consumer should acknowledge delivered messages.
  2931. type AckPolicy int
  2932. const (
  2933. // AckNonePolicy requires no acks for delivered messages.
  2934. AckNonePolicy AckPolicy = iota
  2935. // AckAllPolicy when acking a sequence number, this implicitly acks all
  2936. // sequences below this one as well.
  2937. AckAllPolicy
  2938. // AckExplicitPolicy requires ack or nack for all messages.
  2939. AckExplicitPolicy
  2940. // For configuration mismatch check
  2941. ackPolicyNotSet = 99
  2942. )
  2943. func jsonString(s string) string {
  2944. return "\"" + s + "\""
  2945. }
  2946. func (p *AckPolicy) UnmarshalJSON(data []byte) error {
  2947. switch string(data) {
  2948. case jsonString("none"):
  2949. *p = AckNonePolicy
  2950. case jsonString("all"):
  2951. *p = AckAllPolicy
  2952. case jsonString("explicit"):
  2953. *p = AckExplicitPolicy
  2954. default:
  2955. return fmt.Errorf("nats: can not unmarshal %q", data)
  2956. }
  2957. return nil
  2958. }
  2959. func (p AckPolicy) MarshalJSON() ([]byte, error) {
  2960. switch p {
  2961. case AckNonePolicy:
  2962. return json.Marshal("none")
  2963. case AckAllPolicy:
  2964. return json.Marshal("all")
  2965. case AckExplicitPolicy:
  2966. return json.Marshal("explicit")
  2967. default:
  2968. return nil, fmt.Errorf("nats: unknown acknowlegement policy %v", p)
  2969. }
  2970. }
  2971. func (p AckPolicy) String() string {
  2972. switch p {
  2973. case AckNonePolicy:
  2974. return "AckNone"
  2975. case AckAllPolicy:
  2976. return "AckAll"
  2977. case AckExplicitPolicy:
  2978. return "AckExplicit"
  2979. case ackPolicyNotSet:
  2980. return "Not Initialized"
  2981. default:
  2982. return "Unknown AckPolicy"
  2983. }
  2984. }
  2985. // ReplayPolicy determines how the consumer should replay messages it already has queued in the stream.
  2986. type ReplayPolicy int
  2987. const (
  2988. // ReplayInstantPolicy will replay messages as fast as possible.
  2989. ReplayInstantPolicy ReplayPolicy = iota
  2990. // ReplayOriginalPolicy will maintain the same timing as the messages were received.
  2991. ReplayOriginalPolicy
  2992. // For configuration mismatch check
  2993. replayPolicyNotSet = 99
  2994. )
  2995. func (p *ReplayPolicy) UnmarshalJSON(data []byte) error {
  2996. switch string(data) {
  2997. case jsonString("instant"):
  2998. *p = ReplayInstantPolicy
  2999. case jsonString("original"):
  3000. *p = ReplayOriginalPolicy
  3001. default:
  3002. return fmt.Errorf("nats: can not unmarshal %q", data)
  3003. }
  3004. return nil
  3005. }
  3006. func (p ReplayPolicy) MarshalJSON() ([]byte, error) {
  3007. switch p {
  3008. case ReplayOriginalPolicy:
  3009. return json.Marshal("original")
  3010. case ReplayInstantPolicy:
  3011. return json.Marshal("instant")
  3012. default:
  3013. return nil, fmt.Errorf("nats: unknown replay policy %v", p)
  3014. }
  3015. }
  3016. var (
  3017. ackAck = []byte("+ACK")
  3018. ackNak = []byte("-NAK")
  3019. ackProgress = []byte("+WPI")
  3020. ackTerm = []byte("+TERM")
  3021. )
  3022. // DeliverPolicy determines how the consumer should select the first message to deliver.
  3023. type DeliverPolicy int
  3024. const (
  3025. // DeliverAllPolicy starts delivering messages from the very beginning of a
  3026. // stream. This is the default.
  3027. DeliverAllPolicy DeliverPolicy = iota
  3028. // DeliverLastPolicy will start the consumer with the last sequence
  3029. // received.
  3030. DeliverLastPolicy
  3031. // DeliverNewPolicy will only deliver new messages that are sent after the
  3032. // consumer is created.
  3033. DeliverNewPolicy
  3034. // DeliverByStartSequencePolicy will deliver messages starting from a given
  3035. // sequence.
  3036. DeliverByStartSequencePolicy
  3037. // DeliverByStartTimePolicy will deliver messages starting from a given
  3038. // time.
  3039. DeliverByStartTimePolicy
  3040. // DeliverLastPerSubjectPolicy will start the consumer with the last message
  3041. // for all subjects received.
  3042. DeliverLastPerSubjectPolicy
  3043. // For configuration mismatch check
  3044. deliverPolicyNotSet = 99
  3045. )
  3046. func (p *DeliverPolicy) UnmarshalJSON(data []byte) error {
  3047. switch string(data) {
  3048. case jsonString("all"), jsonString("undefined"):
  3049. *p = DeliverAllPolicy
  3050. case jsonString("last"):
  3051. *p = DeliverLastPolicy
  3052. case jsonString("new"):
  3053. *p = DeliverNewPolicy
  3054. case jsonString("by_start_sequence"):
  3055. *p = DeliverByStartSequencePolicy
  3056. case jsonString("by_start_time"):
  3057. *p = DeliverByStartTimePolicy
  3058. case jsonString("last_per_subject"):
  3059. *p = DeliverLastPerSubjectPolicy
  3060. }
  3061. return nil
  3062. }
  3063. func (p DeliverPolicy) MarshalJSON() ([]byte, error) {
  3064. switch p {
  3065. case DeliverAllPolicy:
  3066. return json.Marshal("all")
  3067. case DeliverLastPolicy:
  3068. return json.Marshal("last")
  3069. case DeliverNewPolicy:
  3070. return json.Marshal("new")
  3071. case DeliverByStartSequencePolicy:
  3072. return json.Marshal("by_start_sequence")
  3073. case DeliverByStartTimePolicy:
  3074. return json.Marshal("by_start_time")
  3075. case DeliverLastPerSubjectPolicy:
  3076. return json.Marshal("last_per_subject")
  3077. default:
  3078. return nil, fmt.Errorf("nats: unknown deliver policy %v", p)
  3079. }
  3080. }
  3081. // RetentionPolicy determines how messages in a set are retained.
  3082. type RetentionPolicy int
  3083. const (
  3084. // LimitsPolicy (default) means that messages are retained until any given limit is reached.
  3085. // This could be one of MaxMsgs, MaxBytes, or MaxAge.
  3086. LimitsPolicy RetentionPolicy = iota
  3087. // InterestPolicy specifies that when all known observables have acknowledged a message it can be removed.
  3088. InterestPolicy
  3089. // WorkQueuePolicy specifies that when the first worker or subscriber acknowledges the message it can be removed.
  3090. WorkQueuePolicy
  3091. )
  3092. // DiscardPolicy determines how to proceed when limits of messages or bytes are
  3093. // reached.
  3094. type DiscardPolicy int
  3095. const (
  3096. // DiscardOld will remove older messages to return to the limits. This is
  3097. // the default.
  3098. DiscardOld DiscardPolicy = iota
  3099. //DiscardNew will fail to store new messages.
  3100. DiscardNew
  3101. )
  3102. const (
  3103. limitsPolicyString = "limits"
  3104. interestPolicyString = "interest"
  3105. workQueuePolicyString = "workqueue"
  3106. )
  3107. func (rp RetentionPolicy) String() string {
  3108. switch rp {
  3109. case LimitsPolicy:
  3110. return "Limits"
  3111. case InterestPolicy:
  3112. return "Interest"
  3113. case WorkQueuePolicy:
  3114. return "WorkQueue"
  3115. default:
  3116. return "Unknown Retention Policy"
  3117. }
  3118. }
  3119. func (rp RetentionPolicy) MarshalJSON() ([]byte, error) {
  3120. switch rp {
  3121. case LimitsPolicy:
  3122. return json.Marshal(limitsPolicyString)
  3123. case InterestPolicy:
  3124. return json.Marshal(interestPolicyString)
  3125. case WorkQueuePolicy:
  3126. return json.Marshal(workQueuePolicyString)
  3127. default:
  3128. return nil, fmt.Errorf("nats: can not marshal %v", rp)
  3129. }
  3130. }
  3131. func (rp *RetentionPolicy) UnmarshalJSON(data []byte) error {
  3132. switch string(data) {
  3133. case jsonString(limitsPolicyString):
  3134. *rp = LimitsPolicy
  3135. case jsonString(interestPolicyString):
  3136. *rp = InterestPolicy
  3137. case jsonString(workQueuePolicyString):
  3138. *rp = WorkQueuePolicy
  3139. default:
  3140. return fmt.Errorf("nats: can not unmarshal %q", data)
  3141. }
  3142. return nil
  3143. }
  3144. func (dp DiscardPolicy) String() string {
  3145. switch dp {
  3146. case DiscardOld:
  3147. return "DiscardOld"
  3148. case DiscardNew:
  3149. return "DiscardNew"
  3150. default:
  3151. return "Unknown Discard Policy"
  3152. }
  3153. }
  3154. func (dp DiscardPolicy) MarshalJSON() ([]byte, error) {
  3155. switch dp {
  3156. case DiscardOld:
  3157. return json.Marshal("old")
  3158. case DiscardNew:
  3159. return json.Marshal("new")
  3160. default:
  3161. return nil, fmt.Errorf("nats: can not marshal %v", dp)
  3162. }
  3163. }
  3164. func (dp *DiscardPolicy) UnmarshalJSON(data []byte) error {
  3165. switch strings.ToLower(string(data)) {
  3166. case jsonString("old"):
  3167. *dp = DiscardOld
  3168. case jsonString("new"):
  3169. *dp = DiscardNew
  3170. default:
  3171. return fmt.Errorf("nats: can not unmarshal %q", data)
  3172. }
  3173. return nil
  3174. }
  3175. // StorageType determines how messages are stored for retention.
  3176. type StorageType int
  3177. const (
  3178. // FileStorage specifies on disk storage. It's the default.
  3179. FileStorage StorageType = iota
  3180. // MemoryStorage specifies in memory only.
  3181. MemoryStorage
  3182. )
  3183. const (
  3184. memoryStorageString = "memory"
  3185. fileStorageString = "file"
  3186. )
  3187. func (st StorageType) String() string {
  3188. switch st {
  3189. case MemoryStorage:
  3190. return "Memory"
  3191. case FileStorage:
  3192. return "File"
  3193. default:
  3194. return "Unknown Storage Type"
  3195. }
  3196. }
  3197. func (st StorageType) MarshalJSON() ([]byte, error) {
  3198. switch st {
  3199. case MemoryStorage:
  3200. return json.Marshal(memoryStorageString)
  3201. case FileStorage:
  3202. return json.Marshal(fileStorageString)
  3203. default:
  3204. return nil, fmt.Errorf("nats: can not marshal %v", st)
  3205. }
  3206. }
  3207. func (st *StorageType) UnmarshalJSON(data []byte) error {
  3208. switch string(data) {
  3209. case jsonString(memoryStorageString):
  3210. *st = MemoryStorage
  3211. case jsonString(fileStorageString):
  3212. *st = FileStorage
  3213. default:
  3214. return fmt.Errorf("nats: can not unmarshal %q", data)
  3215. }
  3216. return nil
  3217. }
  3218. // Length of our hash used for named consumers.
  3219. const nameHashLen = 8
  3220. // Computes a hash for the given `name`.
  3221. func getHash(name string) string {
  3222. sha := sha256.New()
  3223. sha.Write([]byte(name))
  3224. b := sha.Sum(nil)
  3225. for i := 0; i < nameHashLen; i++ {
  3226. b[i] = rdigits[int(b[i]%base)]
  3227. }
  3228. return string(b[:nameHashLen])
  3229. }