Skip to content
This repository

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse code

adds configuration options for misultin port and message handler

  • Loading branch information...
commit 18ecb448bad3a234af492fbeda74f235bf00b705 1 parent a5959b4
Alvaro Videla authored
27 README.md
Source Rendered
... ... @@ -1,6 +1,6 @@
1 1 # RabbitMQ Websockets Plugin #
2 2
3   -This plugin exposes a Websockets for RabbitMQ.
  3 +This plugin exposes Websockets for RabbitMQ.
4 4
5 5 The user connects to the host where RabbitMQ is running, using default port `8080`.
6 6
@@ -20,13 +20,15 @@ Get the `rabbitmq-public-umbrella`
20 20
21 21 Get the [misultin_wrapper](https://github.com/videlalvaro/misultin_wrapper) to support Websockets:
22 22
  23 +Inside the `rabbitmq-public-umbrella` directory do:
  24 +
23 25 $ git clone git://github.com/videlalvaro/misultin_wrapper.git
24 26
25   -Clone this repository:
  27 +Then clone this repository:
26 28
27 29 $ git clone git://github.com/videlalvaro/rabbitmq-websockets.git
28 30
29   -Once you have the code you can move into the `rabbitmq-websockets` and test the plugin with the broker:
  31 +Once you have the code you can move into the `rabbitmq-websockets` directory and test the plugin with the broker:
30 32
31 33 $ make run-in-broker
32 34
@@ -42,6 +44,25 @@ You can publish test messages by calling the following helper function inside th
42 44
43 45 All three parameters are binaries.
44 46
  47 +## Configuration ##
  48 +
  49 +This plugin has two parameters that affect it's behavior:
  50 +
  51 +`misultin_port`: The port Misultin should listen to. Default value is `8080`.
  52 +
  53 +`message_handler`: an Erlang tuple with two atoms like {module, function}. Default value is `{rabbit_websockets_util, basic_handler}`.
  54 +
  55 +You can modify such settings on your `rabbitmq.config` file like this:
  56 +
  57 + [
  58 + {rabbit, [
  59 + ...
  60 + %% list of RabbitMQ options
  61 + ]},
  62 + {rabbitmq_websockets, [ {misultin_port, 8081},
  63 + {message_handler, {my_module, my_function}} ]}
  64 + ].
  65 +
45 66 ## License ##
46 67
47 68 See LICENSE.md
3  ebin/rabbitmq_websockets.app
@@ -10,5 +10,6 @@
10 10 ]},
11 11 {registered, []},
12 12 {mod, {rabbit_websockets, []}},
13   - {env, [{misultin_port, 8080}]},
  13 + {env, [ {misultin_port, 8080},
  14 + {message_handler, {rabbit_websockets_util, basic_handler}} ]},
14 15 {applications, [kernel, stdlib, rabbit, amqp_client]}]}.
7 src/rabbit_websockets_util.erl
... ... @@ -1,6 +1,6 @@
1 1 -module(rabbit_websockets_util).
2 2
3   --export([publish_msg/3]).
  3 +-export([publish_msg/3, basic_handler/1]).
4 4
5 5 -include_lib("amqp_client/include/amqp_client.hrl").
6 6
@@ -9,4 +9,7 @@ publish_msg(Exchange, Msg, RKey) ->
9 9 {ok, Channel} = amqp_connection:open_channel(Connection),
10 10 Publish = #'basic.publish'{exchange = Exchange, routing_key = RKey},
11 11 amqp_channel:call(Channel, Publish, #amqp_msg{payload = term_to_binary(Msg)}),
12   - ok.
  12 + ok.
  13 +
  14 +basic_handler(Msg) ->
  15 + binary_to_term(Msg).
9 src/rabbit_websockets_worker.erl
@@ -10,7 +10,7 @@
10 10
11 11 -record(state, {connection, port}).
12 12 -record(http_state, {req}).
13   --record(websocket_state, {ws, conn, consumer}).
  13 +-record(websocket_state, {ws, conn, consumer, handler}).
14 14
15 15 start_link(Port) ->
16 16 gen_server:start_link({global, ?MODULE}, ?MODULE, [Port], []).
@@ -25,12 +25,13 @@ stop() ->
25 25 init([Port]) ->
26 26 process_flag(trap_exit, true),
27 27 {ok, Connection} = amqp_connection:start(#amqp_params_direct{}),
  28 + {ok, Fun} = application:get_env(message_handler),
28 29
29 30 misultin:start_link([{port, Port},
30 31 {loop, fun(Req) -> handle_http(#http_state{req=Req}) end},
31 32 {ws_loop,
32 33 fun(Ws) ->
33   - handle_websocket(#websocket_state{ws=Ws, conn=Connection})
  34 + handle_websocket(#websocket_state{ws=Ws, conn=Connection, handler=Fun})
34 35 end},
35 36 {ws_autoexit, false}]),
36 37 erlang:monitor(process, misultin),
@@ -101,10 +102,10 @@ css_root() ->
101 102 % Websockets Handling Implementation
102 103 % ----------------------------------
103 104
104   -handle_websocket(#websocket_state{ws=Ws, conn=Connection} = State) ->
  105 +handle_websocket(#websocket_state{ws=Ws, conn=Connection, handler={M, F}} = State) ->
105 106 receive
106 107 {amqp_msgs, Msg} ->
107   - Ws:send(binary_to_term(Msg)),
  108 + Ws:send(M:F(Msg)),
108 109 handle_websocket(State);
109 110 {browser, Data} ->
110 111 {E, R} = parse_data(Data),

0 comments on commit 18ecb44

Please sign in to comment.
Something went wrong with that request. Please try again.