Split out net.ml; create amq.direct exchange
This commit is contained in:
parent
9863c45dee
commit
5380080758
|
@ -365,3 +365,9 @@ let amqp_mainloop conn n =
|
||||||
let start (s, peername) =
|
let start (s, peername) =
|
||||||
Connections.start_connection "amqp" issue_banner
|
Connections.start_connection "amqp" issue_banner
|
||||||
amqp_boot amqp_handler amqp_mainloop (s, peername)
|
amqp_boot amqp_handler amqp_mainloop (s, peername)
|
||||||
|
|
||||||
|
let init () =
|
||||||
|
Node.send_ignore "factory" (Message.create (Sexp.Str "direct",
|
||||||
|
Sexp.Arr [Sexp.Str "amq.direct"],
|
||||||
|
Sexp.Str "", Sexp.Str ""));
|
||||||
|
ignore (Util.create_thread "AMQP listener" None (Net.start_net Amqp_spec.port) start)
|
||||||
|
|
|
@ -0,0 +1,15 @@
|
||||||
|
open Unix
|
||||||
|
|
||||||
|
let rec accept_loop sock connection_start_fn =
|
||||||
|
let (s, peername) = accept sock in
|
||||||
|
setsockopt s TCP_NODELAY true;
|
||||||
|
ignore (connection_start_fn (s, peername));
|
||||||
|
accept_loop sock connection_start_fn
|
||||||
|
|
||||||
|
let start_net port_number connection_start_fn =
|
||||||
|
let sock = socket PF_INET SOCK_STREAM 0 in
|
||||||
|
setsockopt sock SO_REUSEADDR true;
|
||||||
|
bind sock (ADDR_INET (inet_addr_of_string "0.0.0.0", port_number));
|
||||||
|
listen sock 5;
|
||||||
|
Log.info "Accepting connections" [Sexp.Str (string_of_int port_number)];
|
||||||
|
accept_loop sock connection_start_fn
|
25
ocamlmsg.ml
25
ocamlmsg.ml
|
@ -1,21 +1,3 @@
|
||||||
open Unix
|
|
||||||
open Printf
|
|
||||||
open Thread
|
|
||||||
|
|
||||||
let rec accept_loop sock connection_start_fn =
|
|
||||||
let (s, peername) = accept sock in
|
|
||||||
setsockopt s TCP_NODELAY true;
|
|
||||||
ignore (connection_start_fn (s, peername));
|
|
||||||
accept_loop sock connection_start_fn
|
|
||||||
|
|
||||||
let start_net port_number connection_start_fn =
|
|
||||||
let sock = socket PF_INET SOCK_STREAM 0 in
|
|
||||||
setsockopt sock SO_REUSEADDR true;
|
|
||||||
bind sock (ADDR_INET (inet_addr_of_string "0.0.0.0", port_number));
|
|
||||||
listen sock 5;
|
|
||||||
Log.info "Accepting connections" [Sexp.Str (string_of_int port_number)];
|
|
||||||
accept_loop sock connection_start_fn
|
|
||||||
|
|
||||||
let hook_log () =
|
let hook_log () =
|
||||||
let old_hook = !Log.hook in
|
let old_hook = !Log.hook in
|
||||||
let new_hook label body =
|
let new_hook label body =
|
||||||
|
@ -25,12 +7,13 @@ let hook_log () =
|
||||||
Log.hook := new_hook
|
Log.hook := new_hook
|
||||||
|
|
||||||
let _ =
|
let _ =
|
||||||
printf "%s %s, %s %s\n%!" App_info.product App_info.version App_info.copyright App_info.licence;
|
Printf.printf "%s %s, %s %s\n%!"
|
||||||
|
App_info.product App_info.version App_info.copyright App_info.licence;
|
||||||
Sys.set_signal Sys.sigpipe Sys.Signal_ignore;
|
Sys.set_signal Sys.sigpipe Sys.Signal_ignore;
|
||||||
Uuid.init ();
|
Uuid.init ();
|
||||||
Factory.init ();
|
Factory.init ();
|
||||||
Queuenode.init ();
|
Queuenode.init ();
|
||||||
Directnode.init ();
|
Directnode.init ();
|
||||||
hook_log ();
|
hook_log ();
|
||||||
ignore (Util.create_thread "AMQP listener" None (start_net Amqp_spec.port) Amqp_relay.start);
|
Amqp_relay.init ();
|
||||||
start_net 5671 Relay.start
|
Net.start_net 5671 Relay.start
|
||||||
|
|
Loading…
Reference in New Issue