Initial commit.

deps/rabbitmq_random_exchange/.gitignore

@@ -0,0 +1,4 @@
+.DS_Store
+deps
+dist
+ebin

deps/rabbitmq_random_exchange/Makefile

@@ -0,0 +1,37 @@
+PACKAGE=random-exchange
+DIST_DIR=dist
+EBIN_DIR=ebin
+INCLUDE_DIRS=include
+DEPS_DIR=deps
+DEPS ?= 
+DEPS_EZ=$(foreach DEP, $(DEPS), $(DEPS_DIR)/$(DEP).ez)
+RABBITMQ_HOME ?= .
+
+all: compile
+
+clean:
+	rm -rf $(DIST_DIR)
+	rm -rf $(EBIN_DIR)
+
+distclean: clean
+	rm -rf $(DEPS_DIR)
+
+package: compile $(DEPS_EZ)
+	rm -f $(DIST_DIR)/$(PACKAGE).ez
+	mkdir -p $(DIST_DIR)/$(PACKAGE)
+	cp -r $(EBIN_DIR) $(DIST_DIR)/$(PACKAGE)
+	$(foreach EXTRA_DIR, $(INCLUDE_DIRS), cp -r $(EXTRA_DIR) $(DIST_DIR)/$(PACKAGE);)
+	(cd $(DIST_DIR); zip -r $(PACKAGE).ez $(PACKAGE))
+
+install: package
+	$(foreach DEP, $(DEPS_EZ), cp $(DEP) $(RABBITMQ_HOME)/plugins;)
+	cp $(DIST_DIR)/$(PACKAGE).ez $(RABBITMQ_HOME)/plugins
+
+$(DEPS_DIR):
+	./rebar get-deps
+
+$(DEPS_EZ): 
+	cd $(DEPS_DIR); $(foreach DEP, $(DEPS), zip -r $(DEP).ez $(DEP);)
+
+compile: $(DEPS_DIR)
+	./rebar compile

deps/rabbitmq_random_exchange/rebar.config

@@ -0,0 +1,5 @@
+{
+  deps, [
+    {rabbit_common, ".*", {git, "https://github.com/jbrisbin/rabbit_common.git", "HEAD"}}
+  ]
+}.

deps/rabbitmq_random_exchange/src/rabbit_exchange_type_random.erl

@@ -0,0 +1,61 @@
+%% The contents of this file are subject to the Mozilla Public License
+%% Version 1.1 (the "License"); you may not use this file except in
+%% compliance with the License. You may obtain a copy of the License
+%% at http://www.mozilla.org/MPL/
+%%
+%% Software distributed under the License is distributed on an "AS IS"
+%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+%% the License for the specific language governing rights and
+%% limitations under the License.
+%%
+%% The Original Code is RabbitMQ.
+%%
+%% The Initial Developer of the Original Code is VMware, Inc.
+%% Copyright (c) 2007-2011 VMware, Inc.  All rights reserved.
+%%
+
+-module(rabbit_exchange_type_random).
+-include_lib("rabbit_common/include/rabbit.hrl").
+
+-behaviour(rabbit_exchange_type).
+
+-export([description/0, route/2]).
+-export([
+  validate/1, 
+  create/2, 
+  recover/2,
+  delete/3,
+  add_binding/3, 
+  remove_bindings/3, 
+  assert_args_equivalence/2
+]).
+-include_lib("rabbit_common/include/rabbit_exchange_type_spec.hrl").
+
+-rabbit_boot_step({?MODULE,
+                   [{description, "exchange type random"},
+                    {mfa,         {rabbit_registry, register, [exchange, <<"random">>, ?MODULE]}},
+                    {requires,    rabbit_registry},
+                    {enables,     kernel_ready}]}).
+
+description() ->
+    [{name, <<"random">>},
+     {description, <<"AMQP random exchange. Like a direct exchange, but randomly chooses who to route to.">>}].
+
+route(#exchange{name = Name},
+      #delivery{message = #basic_message{routing_keys = Routes}}) ->
+    Matches = rabbit_router:match_routing_key(Name, Routes),
+    case length(Matches) of
+      Len when Len < 2 -> Matches;
+      Len ->
+        Rand = crypto:rand_uniform(1, Len + 1),
+        [lists:nth(Rand, Matches)]
+    end.
+
+validate(_X) -> ok.
+create(_Tx, _X) -> ok.
+recover(_X, _Bs) -> ok.
+delete(_Tx, _X, _Bs) -> ok.
+add_binding(_Tx, _X, _B) -> ok.
+remove_bindings(_Tx, _X, _Bs) -> ok.
+assert_args_equivalence(X, Args) ->
+    rabbit_exchange:assert_args_equivalence(X, Args).

deps/rabbitmq_random_exchange/src/random_exchange.app.src

@@ -0,0 +1,10 @@
+{application, random_exchange,
+ [
+  {description, "RabbitMQ Random Exchange Plugin"},
+  {vsn, "0.1.0"},
+  {modules, []},
+  {registered, []},
+  {env, []},
+  {applications, [kernel, stdlib, rabbit, mnesia]}
+ ]
+}.

deps/rabbitmq_random_exchange/test/random_x_consumer.py

@@ -0,0 +1,25 @@
+#!/usr/bin/env python
+
+import amqplib.client_0_8 as amqp
+
+def callback(msg):
+  print (msg.body)
+  msg.channel.basic_ack(msg.delivery_tag)
+  
+def main():
+  conn = amqp.Connection()
+  channel = conn.channel()
+  exch = channel.exchange_declare("randomtest", "random", auto_delete=False)
+  
+  q, _, _ = channel.queue_declare()
+  channel.queue_bind(q, "randomtest", "random")
+  channel.basic_consume(q, callback=callback)
+  
+  while channel.callbacks:
+    channel.wait()
+    
+  channel.close()
+  conn.close()
+
+if __name__ == '__main__':
+  main()

deps/rabbitmq_random_exchange/test/random_x_publisher.py

@@ -0,0 +1,18 @@
+#!/usr/bin/env python
+
+import amqplib.client_0_8 as amqp
+
+def main():
+  conn = amqp.Connection()
+  channel = conn.channel()
+  exch = channel.exchange_declare("randomtest", "random", auto_delete=False)
+  
+  for i in range(100):
+    msg = amqp.Message("hello world! %s" % i)
+    channel.basic_publish(msg, "randomtest", "random")
+    
+  channel.close()
+  conn.close()
+
+if __name__ == '__main__':
+  main()