Merge pull request #1 from wistia/jz-endpoints

Add methods for nsqd and nsqlookupd HTTP endpoints

Gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    nsq-cluster (0.0.0)
+    nsq-cluster (0.1.1)
       nsq-cluster
 
 GEM

README.md

@@ -15,3 +15,64 @@ cluster.nsqd.last.start
 # Tear down the whole cluster
 cluster.destroy
 ```
+
+Available methods that map to [`nsqd`'s](http://nsq.io/components/nsqd.html) HTTP endpoints.
+
+```ruby
+# nsqd
+nsqd = cluster.nsqd.first
+
+# Publish a message to a topic
+nsqd.pub('stats', 'a message')
+
+# Publish multiple messages to a topic
+nsqd.mpub('stats', 'a message', 'a second message', 'last message')
+
+# Create a topic
+nsqd.create(topic: 'stats')
+
+# Create a channel for a known topic
+nsqd.create(topic: 'stats', channel: 'default')
+
+# Follow the same argument pattern for #delete, #empty, #pause, and #unpause
+
+# Get stats in JSON format
+nsqd.stats
+
+# Ping nsqd
+nsqd.ping
+
+# Get general information
+nsqd.info
+```
+
+Available methods that map to [`nsqlookupd`'s](http://nsq.io/components/nsqlookupd.html) HTTP endpoints.
+
+```ruby
+#nsqlookupd
+nsqlookupd = cluster.nsqlookupd.first
+
+# Look up list of producers by topic
+nsqlookupd.lookup('stats')
+
+# Get a list of known topics
+nsqlookupd.topics
+
+# Get a list of known channels for a topic
+nsqlookupd.channels('stats')
+
+# Get a list of known nodes
+nsqlookupd.nodes
+
+# Delete a topic
+nsqlookupd.delete(topic: 'stats')
+
+# Delete a channel for a known topic
+nsqlookupd.delete(topic: 'stats', channel: 'default')
+
+# Ping nsqlookupd
+nsqlookupd.ping
+
+# Get general info
+nsqlookupd.info
+```

VERSION

@@ -1 +1 @@
-0.1.0
+0.1.1

bin/nsq-cluster

@@ -1,5 +1,7 @@
 #!/usr/bin/env ruby
 
+$:.unshift File.join(File.dirname(__FILE__), "../lib")
+
 require 'nsq-cluster'
 require 'optparse'
 

lib/nsq-cluster/http_wrapper.rb

@@ -0,0 +1,32 @@
+require 'net/http'
+require 'uri'
+
+module HTTPWrapper
+
+
+  def post(path, params = {}, body = nil)
+    uri = uri("#{path}?#{URI.encode_www_form(params)}")
+    request = Net::HTTP::Post.new(uri)
+    request.body = body
+
+    Net::HTTP.start(uri.hostname, uri.port) do |http|
+      http.request(request)
+    end
+  end
+
+
+  def get(path, params = {})
+    uri = uri(path)
+    uri.query = URI.encode_www_form(params)
+    Net::HTTP.get_response(uri)
+  end
+
+
+  private
+
+
+  def uri(path)
+    URI("http://#{@host}:#{@http_port}/#{path}")
+  end
+
+end

lib/nsq-cluster/nsqd.rb

@@ -1,7 +1,9 @@
 require_relative 'process_wrapper'
+require_relative 'http_wrapper'
 require 'fileutils'
 
 class Nsqd < ProcessWrapper
+  include HTTPWrapper
 
 
   attr_reader :host, :tcp_port, :http_port
@@ -60,8 +62,69 @@ def data_path
   end
 
 
+  # publish a single message to a topic
+  def pub(topic, message)
+    post 'pub', { topic: topic }, message
+  end
+
+
+  # publish multiple messages to a topic
+  def mpub(topic, *messages)
+    post 'mpub', { topic: topic }, messages.join("\n")
+  end
+
+
+  # create a topic or a channel in an existing topic
+  def create(params = {})
+    nsqd_post 'create', topic: params[:topic], channel: params[:channel]
+  end
+
+
+  # delete a topic or a channel in an existing topic
+  def delete(params = {})
+    nsqd_post 'delete', topic: params[:topic], channel: params[:channel]
+  end
+
+
+  # empty a topic or a channel in an existing topic
+  def empty(params = {})
+    nsqd_post 'empty', topic: params[:topic], channel: params[:channel]
+  end
+
+
+  # pause a topic or a channel in a topic
+  def pause(params = {})
+    nsqd_post 'pause', topic: params[:topic], channel: params[:channel]
+  end
+
+
+  # unpause a topic or a channel in a topic
+  def unpause(params = {})
+    nsqd_post 'unpause', topic: params[:topic], channel: params[:channel]
+  end
+
+
+  # return stats in json format
+  def stats
+    get 'stats', format: 'json'
+  end
+
+
+  # monitoring endpoint
+  def ping
+    get 'ping'
+  end
+
+
+  # returns version number
+  def info
+    get 'info'
+  end
+
+
   private
 
+
   def create_data_directory
     Dir.mkdir(data_path)
   end
@@ -71,4 +134,15 @@ def clear_data_directory
     FileUtils.rm_rf(data_path) if Dir.exist?(data_path)
   end
 
+
+  def nsqd_post(action, params)
+    if params[:topic] && params[:channel]
+      post "#{action}_channel", topic: params[:topic], channel: params[:channel]
+    elsif params[:topic]
+      post "#{action}_topic", topic: params[:topic]
+    else
+      raise 'you must specify a topic or topic and channel'
+    end
+  end
+
 end

lib/nsq-cluster/nsqlookupd.rb

@@ -1,6 +1,8 @@
 require_relative 'process_wrapper'
+require_relative 'http_wrapper'
 
 class Nsqlookupd < ProcessWrapper
+  include HTTPWrapper
 
   attr_reader :host, :tcp_port, :http_port
 
@@ -25,4 +27,60 @@ def args
     ]
   end
 
+
+ # return a list of producers for a topic
+  def lookup(topic)
+    get 'lookup', topic: topic
+  end
+
+
+  # return a list of all known topics
+  def topics
+    get 'topics'
+  end
+
+
+  # return a list of all known channels for a topic
+  def channels(topic)
+    get 'channels', topic: topic
+  end
+
+
+  # return a list of all known nsqd
+  def nodes
+    get 'nodes'
+  end
+
+
+  # delete a topic or a channel in an existing topic
+  def delete(params = {})
+    nsqlookupd_post 'delete', topic: params[:topic], channel: params[:channel]
+  end
+
+
+  # monitoring endpoint
+  def ping
+    get 'ping'
+  end
+
+
+  # returns version number
+  def info
+    get 'info'
+  end
+
+
+  private
+
+
+  def nsqlookupd_post(action, params)
+    if params[:topic] && params[:channel]
+      post "#{action}_channel", topic: params[:topic], channel: params[:channel]
+    elsif params[:topic]
+      post "#{action}_topic", topic: params[:topic]
+    else
+      raise 'you must specify a topic or topic and channel'
+    end
+  end
+
 end

nsq-cluster.gemspec

@@ -2,16 +2,16 @@
 # DO NOT EDIT THIS FILE DIRECTLY
 # Instead, edit Jeweler::Tasks in Rakefile, and run 'rake gemspec'
 # -*- encoding: utf-8 -*-
-# stub: nsq-cluster 0.1.0 ruby lib
+# stub: nsq-cluster 0.1.1 ruby lib
 
 Gem::Specification.new do |s|
   s.name = "nsq-cluster"
-  s.version = "0.1.0"
+  s.version = "0.1.1"
 
   s.required_rubygems_version = Gem::Requirement.new(">= 0") if s.respond_to? :required_rubygems_version=
   s.require_paths = ["lib"]
   s.authors = ["Brendan Schwartz"]
-  s.date = "2014-04-14"
+  s.date = "2014-06-26"
   s.description = "Setup nsqd, nsqlookupd, and nsqadmin in a jiffy. Great for testing!"
   s.email = "[email protected]"
   s.executables = ["nsq-cluster"]
@@ -31,13 +31,16 @@ Gem::Specification.new do |s|
     "VERSION",
     "bin/nsq-cluster",
     "lib/nsq-cluster.rb",
+    "lib/nsq-cluster/http_wrapper.rb",
     "lib/nsq-cluster/nsqadmin.rb",
     "lib/nsq-cluster/nsqd.rb",
     "lib/nsq-cluster/nsqlookupd.rb",
     "lib/nsq-cluster/process_wrapper.rb",
     "nsq-cluster.gemspec",
     "test/helper.rb",
-    "test/nsq_cluster_spec.rb"
+    "test/nsq_cluster_spec.rb",
+    "test/nsqd_spec.rb",
+    "test/nsqlookupd_spec.rb"
   ]
   s.homepage = "http://github.com/wistia/nsq-cluster"
   s.licenses = ["MIT"]

test/helper.rb

@@ -1,3 +1,4 @@
+require 'json'
 require 'simplecov'
 require 'minitest/autorun'
 require 'minitest/pride'