Module: NATS::JetStream::Msg::AckMethods

Included in:
Msg
Defined in:
lib/nats/io/jetstream/msg/ack_methods.rb

Instance Method Summary collapse

Instance Method Details

#ack(**params) ⇒ Object 



19
20
21
22
23
24
25
26
27
28
29
30
 
# File 'lib/nats/io/jetstream/msg/ack_methods.rb', line 19

def ack(**params)
  ensure_is_acked_once!

  resp = if params[:timeout]
           @nc.request(@reply, Ack::Ack, **params)
         else
           @nc.publish(@reply, Ack::Ack)
         end
  @sub.synchronize { @ackd = true }

  resp
end

#ack_sync(**params) ⇒ Object 



32
33
34
35
36
37
38
39
40
 
# File 'lib/nats/io/jetstream/msg/ack_methods.rb', line 32

def ack_sync(**params)
  ensure_is_acked_once!

  params[:timeout] ||= 0.5
  resp = @nc.request(@reply, Ack::Ack, **params)
  @sub.synchronize { @ackd = true }

  resp
end

#in_progress(**params) ⇒ Object 



72
73
74
 
# File 'lib/nats/io/jetstream/msg/ack_methods.rb', line 72

def in_progress(**params)
  params[:timeout] ? @nc.request(@reply, Ack::Progress, **params) : @nc.publish(@reply, Ack::Progress)
end

#metadata ⇒ Object 



76
77
78
 
# File 'lib/nats/io/jetstream/msg/ack_methods.rb', line 76

def metadata
  @meta ||= parse_metadata(reply)
end

#nak(**params) ⇒ Object 



42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/nats/io/jetstream/msg/ack_methods.rb', line 42

def nak(**params)
  ensure_is_acked_once!
  payload = if params[:delay]
              payload = "#{Ack::Nak} #{{ delay: params[:delay] }.to_json}"
            else
              Ack::Nak
            end
  resp = if params[:timeout]
           @nc.request(@reply, payload, **params)
         else
           @nc.publish(@reply, payload)
         end
  @sub.synchronize { @ackd = true }

  resp
end

#term(**params) ⇒ Object 



59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/nats/io/jetstream/msg/ack_methods.rb', line 59

def term(**params)
  ensure_is_acked_once!

  resp = if params[:timeout]
           @nc.request(@reply, Ack::Term, **params)
         else
           @nc.publish(@reply, Ack::Term)
         end
  @sub.synchronize { @ackd = true }

  resp
end