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 



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

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 



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

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 



74
75
76
 
# File 'lib/nats/io/jetstream/msg/ack_methods.rb', line 74

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

#metadata ⇒ Object 



78
79
80
 
# File 'lib/nats/io/jetstream/msg/ack_methods.rb', line 78

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

#nak(**params) ⇒ Object 



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

def nak(**params)
  ensure_is_acked_once!
  payload = if params[:delay]
    "#{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 



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

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