Class: Nsqd

Inherits:
ProcessWrapper show all
Includes:
HTTPWrapper
Defined in:
lib/nsq-cluster/nsqd.rb

Constant Summary

Constants inherited from ProcessWrapper

ProcessWrapper::HTTPCHECK_INTERVAL

Instance Attribute Summary collapse

Attributes inherited from ProcessWrapper

#pid

Class Method Summary collapse

Instance Method Summary collapse

Methods included from HTTPWrapper

#get, #post

Methods inherited from ProcessWrapper

#another_instance_is_running?, #block_until_running, #block_until_stopped, #output, #running?, #start, #stop

Constructor Details

#initialize(opts = {}, verbose = false) ⇒ Nsqd

Returns a new instance of Nsqd.



19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/nsq-cluster/nsqd.rb', line 19

def initialize(opts = {}, verbose = false)
  super

  @id = opts.delete(:id) || 0
  @host = opts.delete(:host) || '127.0.0.1'

  # Use a non-standard nsqd port by default so as to not conflict with any
  # local instances. This is helpful when running tests!
  @base_port = opts.delete(:base_port) || 4250

  @tcp_port = opts.delete(:tcp_port) || (@base_port + @id * 2)
  @http_port = opts.delete(:http_port) || (@base_port + 1 + @id * 2)
  @lookupd = opts.delete(:nsqlookupd) || []
  @broadcast_address = opts.delete(:broadcast_address) || @host

  @extra_args = opts.map do |key, value|
    "--#{key.to_s.gsub('_', '-')}=#{value}"
  end

  clear_data_directory
  create_data_directory
end

Instance Attribute Details

#base_port ⇒ Object (readonly)

Returns the value of attribute base_port.



9
10
11
 
# File 'lib/nsq-cluster/nsqd.rb', line 9

def base_port
  @base_port
end

#host ⇒ Object (readonly)

Returns the value of attribute host.



9
10
11
 
# File 'lib/nsq-cluster/nsqd.rb', line 9

def host
  @host
end

#http_port ⇒ Object (readonly)

Returns the value of attribute http_port.



9
10
11
 
# File 'lib/nsq-cluster/nsqd.rb', line 9

def http_port
  @http_port
end

#id ⇒ Object (readonly)

Returns the value of attribute id.



9
10
11
 
# File 'lib/nsq-cluster/nsqd.rb', line 9

def id
  @id
end

#tcp_port ⇒ Object (readonly)

Returns the value of attribute tcp_port.



9
10
11
 
# File 'lib/nsq-cluster/nsqd.rb', line 9

def tcp_port
  @tcp_port
end

Class Method Details

.version_is_pre_1? ⇒ Boolean

Returns true if nsqd's version is < 1.0

Returns:

  • (Boolean) 


12
13
14
15
16
 
# File 'lib/nsq-cluster/nsqd.rb', line 12

def self.version_is_pre_1?
  @version_is_pre_1 ||= (
    `nsqd -version`.index('nsqd v0') == 0
  )
end

Instance Method Details

#args ⇒ Object 



54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/nsq-cluster/nsqd.rb', line 54

def args
  base_args = [
    %Q(--tcp-address=#{@host}:#{@tcp_port}),
    %Q(--http-address=#{@host}:#{@http_port}),
    %Q(--data-path=#{data_path}),
    %Q(--broadcast-address=#{@broadcast_address})
  ]

  if Nsqd.version_is_pre_1?
    node_args = [%Q(--worker-id=#{id})]
  else
    node_args = [%Q(--node-id=#{id})]
  end

  lookupd_args = @lookupd.map do |ld|
    %Q(--lookupd-tcp-address=#{ld.host}:#{ld.tcp_port})
  end

  base_args + node_args + @extra_args + lookupd_args
end

#command ⇒ Object 



49
50
51
 
# File 'lib/nsq-cluster/nsqd.rb', line 49

def command
  'nsqd'
end

#create(params = {}) ⇒ Object

create a topic or a channel in an existing topic



95
96
97
 
# File 'lib/nsq-cluster/nsqd.rb', line 95

def create(params = {})
  nsqd_post 'create', topic: params[:topic], channel: params[:channel]
end

#data_path ⇒ Object

find or create a temporary data directory for this instance



77
78
79
 
# File 'lib/nsq-cluster/nsqd.rb', line 77

def data_path
  "/tmp/nsqd-#{id}"
end

#delete(params = {}) ⇒ Object

delete a topic or a channel in an existing topic



101
102
103
 
# File 'lib/nsq-cluster/nsqd.rb', line 101

def delete(params = {})
  nsqd_post 'delete', topic: params[:topic], channel: params[:channel]
end

#destroy ⇒ Object 



43
44
45
46
 
# File 'lib/nsq-cluster/nsqd.rb', line 43

def destroy
  super
  clear_data_directory
end

#empty(params = {}) ⇒ Object

empty a topic or a channel in an existing topic



107
108
109
 
# File 'lib/nsq-cluster/nsqd.rb', line 107

def empty(params = {})
  nsqd_post 'empty', topic: params[:topic], channel: params[:channel]
end

#info ⇒ Object

returns version number



137
138
139
 
# File 'lib/nsq-cluster/nsqd.rb', line 137

def info
  get 'info'
end

#mpub(topic, *messages) ⇒ Object

publish multiple messages to a topic



89
90
91
 
# File 'lib/nsq-cluster/nsqd.rb', line 89

def mpub(topic, *messages)
  post 'mpub', { topic: topic }, messages.join("\n")
end

#pause(params = {}) ⇒ Object

pause a topic or a channel in a topic



113
114
115
 
# File 'lib/nsq-cluster/nsqd.rb', line 113

def pause(params = {})
  nsqd_post 'pause', topic: params[:topic], channel: params[:channel]
end

#ping ⇒ Object

monitoring endpoint



131
132
133
 
# File 'lib/nsq-cluster/nsqd.rb', line 131

def ping
  get 'ping'
end

#pub(topic, message) ⇒ Object

publish a single message to a topic



83
84
85
 
# File 'lib/nsq-cluster/nsqd.rb', line 83

def pub(topic, message)
  post 'pub', { topic: topic }, message
end

#stats ⇒ Object

return stats in json format



125
126
127
 
# File 'lib/nsq-cluster/nsqd.rb', line 125

def stats
  get 'stats', format: 'json'
end

#unpause(params = {}) ⇒ Object

unpause a topic or a channel in a topic



119
120
121
 
# File 'lib/nsq-cluster/nsqd.rb', line 119

def unpause(params = {})
  nsqd_post 'unpause', topic: params[:topic], channel: params[:channel]
end