Merge default into bug22825

12 files changed, 598 insertions, 120 deletions

diff --git a/tools/Makefile.am b/tools/Makefile.am
index 2c47385..ad53d88 100644
--- a/tools/Makefile.am
+++ b/tools/Makefile.am
@@ -2,15 +2,21 @@ SUBDIRS=doc
 
 bin_PROGRAMS = amqp-publish amqp-get amqp-consume amqp-declare-queue amqp-delete-queue
 
-AM_CFLAGS = -I$(top_srcdir)/librabbitmq
+AM_CFLAGS = -I$(top_srcdir)/librabbitmq -I$(PLATFORM_DIR)
 AM_LDFLAGS = $(top_builddir)/librabbitmq/librabbitmq.la
 
 LDADD=$(LIBPOPT)
 
-noinst_HEADERS = common.h
+noinst_HEADERS = common.h $(PLATFORM_DIR)/process.h
 
-amqp_publish_SOURCES = publish.c common.c
-amqp_get_SOURCES = get.c common.c
-amqp_consume_SOURCES = consume.c common.c
-amqp_declare_queue_SOURCES = declare_queue.c common.c
-amqp_delete_queue_SOURCES = delete_queue.c common.c
+COMMON_SOURCES = common.c
+
+if WINDOWS
+COMMON_SOURCES += windows/compat.c
+endif
+
+amqp_publish_SOURCES = publish.c $(COMMON_SOURCES)
+amqp_get_SOURCES = get.c $(COMMON_SOURCES)
+amqp_consume_SOURCES = consume.c $(PLATFORM_DIR)/process.c $(COMMON_SOURCES)
+amqp_declare_queue_SOURCES = declare_queue.c $(COMMON_SOURCES)
+amqp_delete_queue_SOURCES = delete_queue.c $(COMMON_SOURCES)
diff --git a/tools/common.c b/tools/common.c
index 6a38a95..39099c1 100644
--- a/tools/common.c
+++ b/tools/common.c
@@ -58,14 +58,12 @@
 #include <unistd.h>
 #include <fcntl.h>
 #include <errno.h>
-#include <spawn.h>
-#include <sys/wait.h>
-
-#include <popt.h>
 
 #include "common.h"
 
-extern char **environ;
+#ifdef WINDOWS
+#include "compat.h"
+#endif
 
 void die(const char *fmt, ...)
 {
@@ -86,11 +84,24 @@ void die_errno(int err, const char *fmt, ...)
 	va_start(ap, fmt);
 	vfprintf(stderr, fmt, ap);
 	va_end(ap);
-	fprintf(stderr, ": %s\n", strerror(err));
+	fprintf(stderr, ": %s\n", strerror(errno));
 	exit(1);
 }
 
-char *amqp_server_exception_string(amqp_rpc_reply_t r)
+void die_amqp_error(int err, const char *fmt, ...)
+{
+	if (err <= 0)
+		return;
+	
+	va_list ap;
+	va_start(ap, fmt);
+	vfprintf(stderr, fmt, ap);
+	va_end(ap);
+	fprintf(stderr, ": %s\n", amqp_error_string(err));
+	exit(1);
+}
+
+const char *amqp_server_exception_string(amqp_rpc_reply_t r)
 {
 	int res;
 	char *s;
@@ -125,26 +136,17 @@ char *amqp_server_exception_string(amqp_rpc_reply_t r)
 	return res >= 0 ? s : NULL;
 }
 
-char *amqp_rpc_reply_string(amqp_rpc_reply_t r)
+const char *amqp_rpc_reply_string(amqp_rpc_reply_t r)
 {
-	const char *s;
-
 	switch (r.reply_type) {
 	case AMQP_RESPONSE_NORMAL:
-		s = "normal response";
-		break;
+		return strdup("normal response");
 		
 	case AMQP_RESPONSE_NONE:
-		s = "missing RPC reply type";
-		break;
+		return strdup("missing RPC reply type");
 
 	case AMQP_RESPONSE_LIBRARY_EXCEPTION:
-		if (r.library_errno)
-			s = strerror(r.library_errno);
-		else
-			s = "end of stream";
-
-		break;
+		return amqp_error_string(r.library_error);
 
 	case AMQP_RESPONSE_SERVER_EXCEPTION:
 		return amqp_server_exception_string(r);
@@ -152,8 +154,6 @@ char *amqp_rpc_reply_string(amqp_rpc_reply_t r)
 	default:
 		abort();
 	}
-
-	return strdup(s);
 }
 
 void die_rpc(amqp_rpc_reply_t r, const char *fmt, ...)
@@ -169,16 +169,6 @@ void die_rpc(amqp_rpc_reply_t r, const char *fmt, ...)
 	exit(1);
 }
 
-void set_cloexec(int fd)
-{
-	int flags;
-
-	flags = fcntl(fd, F_GETFD);
-	if (flags == -1
-	    || fcntl(fd, F_SETFD, (long)(flags | FD_CLOEXEC)) == -1)
-		die_errno(errno, "set_cloexec");
-}
-
 static char *amqp_server = "localhost";
 static char *amqp_vhost = "/";
 static char *amqp_username = "guest";
@@ -222,14 +212,7 @@ amqp_connection_state_t make_connection(void)
 	}
 
 	s = amqp_open_socket(host, port ? port : 5672);
-	if (s < 0) {
-		if (s == -ENOENT)
-			die("unknown host %s", host);
-		else
-			die_errno(-s, "opening socket to %s", amqp_server);
-	}
-
-	set_cloexec(s);
+	die_amqp_error(-s, "opening socket to %s", amqp_server);
 	
 	conn = amqp_new_connection();
 	amqp_set_sockfd(conn, s);
@@ -247,16 +230,14 @@ amqp_connection_state_t make_connection(void)
 
 void close_connection(amqp_connection_state_t conn)
 {
-	int s = amqp_get_sockfd(conn);
-
+	int res;
 	die_rpc(amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS),
 		"closing channel");
 	die_rpc(amqp_connection_close(conn, AMQP_REPLY_SUCCESS),
 		"closing connection");
-	amqp_destroy_connection(conn);
-
-	if (close(s) < 0)
-		die_errno(errno, "closing socket");
+	
+	res = amqp_end_connection(conn);
+	die_amqp_error(-res, "closing connection");
 }
 
 amqp_bytes_t read_all(int fd)
@@ -307,8 +288,7 @@ void copy_body(amqp_connection_state_t conn, int fd)
 	amqp_frame_t frame;
 		
 	int res = amqp_simple_wait_frame(conn, &frame);
-	if (res < 0)
-		die_errno(-res, "waiting for header frame");
+	die_amqp_error(-res, "waiting for header frame");
 	if (frame.frame_type != AMQP_FRAME_HEADER)
 		die("expected header, got frame type 0x%X",
 		    frame.frame_type);
@@ -316,8 +296,7 @@ void copy_body(amqp_connection_state_t conn, int fd)
 	body_remaining = frame.payload.properties.body_size;
 	while (body_remaining) {
 		res = amqp_simple_wait_frame(conn, &frame);
-		if (res < 0)
-			die_errno(-res, "waiting for body frame");
+		die_amqp_error(-res, "waiting for body frame");
 		if (frame.frame_type != AMQP_FRAME_BODY)
 			die("expected body, got frame type 0x%X",
 			    frame.frame_type);
@@ -327,47 +306,6 @@ void copy_body(amqp_connection_state_t conn, int fd)
 	}
 }
 
-void pipeline(const char * const *argv, struct pipeline *pl)
-{
-	posix_spawn_file_actions_t file_acts;
-
-	int pipefds[2];
-	if (pipe(pipefds))
-		die_errno(errno, "pipe");
-
-	die_errno(posix_spawn_file_actions_init(&file_acts),
-		  "posix_spawn_file_actions_init");
-	die_errno(posix_spawn_file_actions_adddup2(&file_acts, pipefds[0], 0),
-		  "posix_spawn_file_actions_adddup2");
-	die_errno(posix_spawn_file_actions_addclose(&file_acts, pipefds[0]),
-		  "posix_spawn_file_actions_addclose");
-	die_errno(posix_spawn_file_actions_addclose(&file_acts, pipefds[1]),
-		  "posix_spawn_file_actions_addclose");
-
-	die_errno(posix_spawnp(&pl->pid, argv[0], &file_acts, NULL,
-			       (char * const *)argv, environ),
-		  "posix_spawnp: %s", argv[0]);
-
-	die_errno(posix_spawn_file_actions_destroy(&file_acts),
-		  "posix_spawn_file_actions_destroy");
-	
-	if (close(pipefds[0]))
-		die_errno(errno, "close");
-
-	pl->infd = pipefds[1];
-}
-
-int finish_pipeline(struct pipeline *pl)
-{
-	int status;
-
-	if (close(pl->infd))
-		die_errno(errno, "close");
-	if (waitpid(pl->pid, &status, 0) < 0)
-		die_errno(errno, "waitpid");
-	return WIFEXITED(status) && WEXITSTATUS(status) == 0;
-}
-
 poptContext process_options(int argc, const char **argv,
 			    struct poptOption *options,
 			    const char *help)
diff --git a/tools/common.h b/tools/common.h
index 09a9242..0caee98 100644
--- a/tools/common.h
+++ b/tools/common.h
@@ -50,16 +50,20 @@
 
 #include <stdint.h>
 
+#include <popt.h>
+
 #include <amqp.h>
 #include <amqp_framing.h>
 
-extern char *amqp_server_exception_string(amqp_rpc_reply_t r);
-extern char *amqp_rpc_reply_string(amqp_rpc_reply_t r);
+extern const char *amqp_server_exception_string(amqp_rpc_reply_t r);
+extern const char *amqp_rpc_reply_string(amqp_rpc_reply_t r);
 
 extern void die(const char *fmt, ...)
 	__attribute__ ((format (printf, 1, 2)));
 extern void die_errno(int err, const char *fmt, ...)
 	__attribute__ ((format (printf, 2, 3)));
+extern void die_amqp_error(int err, const char *fmt, ...)
+	__attribute__ ((format (printf, 2, 3)));
 extern void die_rpc(amqp_rpc_reply_t r, const char *fmt, ...)
 	__attribute__ ((format (printf, 2, 3)));
 
@@ -73,14 +77,6 @@ extern void write_all(int fd, amqp_bytes_t data);
 
 extern void copy_body(amqp_connection_state_t conn, int fd);
 
-struct pipeline {
-	int pid;
-	int infd;
-};
-
-extern void pipeline(const char * const *argv, struct pipeline *pl);
-extern int finish_pipeline(struct pipeline *pl);
-
 #define INCLUDE_OPTIONS(options) \
 	{NULL, 0, POPT_ARG_INCLUDE_TABLE, options, 0, options ## _title, NULL}
 
diff --git a/tools/consume.c b/tools/consume.c
index 40b61d1..2117bba 100644
--- a/tools/consume.c
+++ b/tools/consume.c
@@ -53,9 +53,8 @@
 #include <stdio.h>
 #include <stdlib.h>
 
-#include <popt.h>
-
 #include "common.h"
+#include "process.h"
 
 /* Convert a amqp_bytes_t to an escaped string form for printing.  We
    use the same escaping conventions as rabbitmqctl. */
@@ -157,8 +156,7 @@ static void do_consume(amqp_connection_state_t conn, amqp_bytes_t queue,
 		struct pipeline pl;
 		uint64_t delivery_tag;
 		int res = amqp_simple_wait_frame(conn, &frame);
-		if (res < 0)
-			die_errno(-res, "waiting for header frame");
+		die_amqp_error(res, "waiting for header frame");
 
 		if (frame.frame_type != AMQP_FRAME_METHOD
 		    || frame.payload.method.id != AMQP_BASIC_DELIVER_METHOD)
@@ -172,8 +170,9 @@ static void do_consume(amqp_connection_state_t conn, amqp_bytes_t queue,
 		copy_body(conn, pl.infd);
 
 		if (finish_pipeline(&pl) && !no_ack)
-			die_errno(-amqp_basic_ack(conn, 1, delivery_tag, 0),
-				  "basic.ack");
+			die_amqp_error(amqp_basic_ack(conn, 1, delivery_tag,
+						      0),
+				       "basic.ack");
 
 		amqp_maybe_release_buffers(conn);
 	}
diff --git a/tools/get.c b/tools/get.c
index f746fd1..8f8e0d0 100644
--- a/tools/get.c
+++ b/tools/get.c
@@ -52,8 +52,6 @@
 
 #include <stdio.h>
 
-#include <popt.h>
-
 #include "common.h"
 
 static int do_get(amqp_connection_state_t conn, char *queue)
diff --git a/tools/publish.c b/tools/publish.c
index 21314b2..0917dae 100644
--- a/tools/publish.c
+++ b/tools/publish.c
@@ -54,8 +54,6 @@
 #include <stdlib.h>
 #include <string.h>
 
-#include <popt.h>
-
 #include "common.h"
 
 static void do_publish(amqp_connection_state_t conn,
@@ -66,8 +64,7 @@ static void do_publish(amqp_connection_state_t conn,
 				     cstring_bytes(exchange),
 				     cstring_bytes(routing_key),
 				     0, 0, props, body);
-	if (res != 0)
-		die_errno(-res, "basic.publish");
+	die_amqp_error(res, "basic.publish");
 }
 
 int main(int argc, const char **argv)
diff --git a/tools/unix/process.c b/tools/unix/process.c
new file mode 100644
index 0000000..8a02afb
--- /dev/null
+++ b/tools/unix/process.c
@@ -0,0 +1,100 @@
+/*
+ * ***** BEGIN LICENSE BLOCK *****
+ * Version: MPL 1.1/GPL 2.0
+ *
+ * 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 librabbitmq.
+ *
+ * The Initial Developers of the Original Code are LShift Ltd, Cohesive
+ * Financial Technologies LLC, and Rabbit Technologies Ltd.  Portions
+ * created before 22-Nov-2008 00:00:00 GMT by LShift Ltd, Cohesive
+ * Financial Technologies LLC, or Rabbit Technologies Ltd are Copyright
+ * (C) 2007-2008 LShift Ltd, Cohesive Financial Technologies LLC, and
+ * Rabbit Technologies Ltd.
+ *
+ * Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
+ * Ltd. Portions created by Cohesive Financial Technologies LLC are
+ * Copyright (C) 2007-2010 Cohesive Financial Technologies
+ * LLC. Portions created by Rabbit Technologies Ltd are Copyright (C)
+ * 2007-2010 Rabbit Technologies Ltd.
+ *
+ * Portions created by Tony Garnock-Jones are Copyright (C) 2009-2010
+ * LShift Ltd and Tony Garnock-Jones.
+ *
+ * All Rights Reserved.
+ *
+ * Contributor(s): ______________________________________.
+ *
+ * Alternatively, the contents of this file may be used under the terms
+ * of the GNU General Public License Version 2 or later (the "GPL"), in
+ * which case the provisions of the GPL are applicable instead of those
+ * above. If you wish to allow use of your version of this file only
+ * under the terms of the GPL, and not to allow others to use your
+ * version of this file under the terms of the MPL, indicate your
+ * decision by deleting the provisions above and replace them with the
+ * notice and other provisions required by the GPL. If you do not
+ * delete the provisions above, a recipient may use your version of
+ * this file under the terms of any one of the MPL or the GPL.
+ *
+ * ***** END LICENSE BLOCK *****
+ */
+
+#include <unistd.h>
+#include <errno.h>
+#include <spawn.h>
+#include <sys/wait.h>
+
+#include "common.h"
+#include "process.h"
+
+extern char **environ;
+
+void pipeline(const char *const *argv, struct pipeline *pl)
+{
+	posix_spawn_file_actions_t file_acts;
+
+	int pipefds[2];
+	if (pipe(pipefds))
+		die_errno(errno, "pipe");
+
+	die_errno(posix_spawn_file_actions_init(&file_acts),
+		  "posix_spawn_file_actions_init");
+	die_errno(posix_spawn_file_actions_adddup2(&file_acts, pipefds[0], 0),
+		  "posix_spawn_file_actions_adddup2");
+	die_errno(posix_spawn_file_actions_addclose(&file_acts, pipefds[0]),
+		  "posix_spawn_file_actions_addclose");
+	die_errno(posix_spawn_file_actions_addclose(&file_acts, pipefds[1]),
+		  "posix_spawn_file_actions_addclose");
+
+	die_errno(posix_spawnp(&pl->pid, argv[0], &file_acts, NULL,
+			       (char * const *)argv, environ),
+		  "posix_spawnp: %s", argv[0]);
+
+	die_errno(posix_spawn_file_actions_destroy(&file_acts),
+		  "posix_spawn_file_actions_destroy");
+	
+	if (close(pipefds[0]))
+		die_errno(errno, "close");
+
+	pl->infd = pipefds[1];
+}
+
+int finish_pipeline(struct pipeline *pl)
+{
+	int status;
+
+	if (close(pl->infd))
+		die_errno(errno, "close");
+	if (waitpid(pl->pid, &status, 0) < 0)
+		die_errno(errno, "waitpid");
+	return WIFEXITED(status) && WEXITSTATUS(status) == 0;
+}
diff --git a/tools/unix/process.h b/tools/unix/process.h
new file mode 100644
index 0000000..ac2939d
--- /dev/null
+++ b/tools/unix/process.h
@@ -0,0 +1,57 @@
+/*
+ * ***** BEGIN LICENSE BLOCK *****
+ * Version: MPL 1.1/GPL 2.0
+ *
+ * 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 librabbitmq.
+ *
+ * The Initial Developers of the Original Code are LShift Ltd, Cohesive
+ * Financial Technologies LLC, and Rabbit Technologies Ltd.  Portions
+ * created before 22-Nov-2008 00:00:00 GMT by LShift Ltd, Cohesive
+ * Financial Technologies LLC, or Rabbit Technologies Ltd are Copyright
+ * (C) 2007-2008 LShift Ltd, Cohesive Financial Technologies LLC, and
+ * Rabbit Technologies Ltd.
+ *
+ * Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
+ * Ltd. Portions created by Cohesive Financial Technologies LLC are
+ * Copyright (C) 2007-2010 Cohesive Financial Technologies
+ * LLC. Portions created by Rabbit Technologies Ltd are Copyright (C)
+ * 2007-2010 Rabbit Technologies Ltd.
+ *
+ * Portions created by Tony Garnock-Jones are Copyright (C) 2009-2010
+ * LShift Ltd and Tony Garnock-Jones.
+ *
+ * All Rights Reserved.
+ *
+ * Contributor(s): ______________________________________.
+ *
+ * Alternatively, the contents of this file may be used under the terms
+ * of the GNU General Public License Version 2 or later (the "GPL"), in
+ * which case the provisions of the GPL are applicable instead of those
+ * above. If you wish to allow use of your version of this file only
+ * under the terms of the GPL, and not to allow others to use your
+ * version of this file under the terms of the MPL, indicate your
+ * decision by deleting the provisions above and replace them with the
+ * notice and other provisions required by the GPL. If you do not
+ * delete the provisions above, a recipient may use your version of
+ * this file under the terms of any one of the MPL or the GPL.
+ *
+ * ***** END LICENSE BLOCK *****
+ */
+
+struct pipeline {
+	int pid;
+	int infd;
+};
+
+extern void pipeline(const char *const *argv, struct pipeline *pl);
+extern int finish_pipeline(struct pipeline *pl);
diff --git a/tools/windows/compat.c b/tools/windows/compat.c
new file mode 100644
index 0000000..f0508b2
--- /dev/null
+++ b/tools/windows/compat.c
@@ -0,0 +1,73 @@
+/*
+ * ***** BEGIN LICENSE BLOCK *****
+ * Version: MPL 1.1/GPL 2.0
+ *
+ * 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 librabbitmq.
+ *
+ * The Initial Developers of the Original Code are LShift Ltd, Cohesive
+ * Financial Technologies LLC, and Rabbit Technologies Ltd.  Portions
+ * created before 22-Nov-2008 00:00:00 GMT by LShift Ltd, Cohesive
+ * Financial Technologies LLC, or Rabbit Technologies Ltd are Copyright
+ * (C) 2007-2008 LShift Ltd, Cohesive Financial Technologies LLC, and
+ * Rabbit Technologies Ltd.
+ *
+ * Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
+ * Ltd. Portions created by Cohesive Financial Technologies LLC are
+ * Copyright (C) 2007-2010 Cohesive Financial Technologies
+ * LLC. Portions created by Rabbit Technologies Ltd are Copyright (C)
+ * 2007-2010 Rabbit Technologies Ltd.
+ *
+ * Portions created by Tony Garnock-Jones are Copyright (C) 2009-2010
+ * LShift Ltd and Tony Garnock-Jones.
+ *
+ * All Rights Reserved.
+ *
+ * Contributor(s): ______________________________________.
+ *
+ * Alternatively, the contents of this file may be used under the terms
+ * of the GNU General Public License Version 2 or later (the "GPL"), in
+ * which case the provisions of the GPL are applicable instead of those
+ * above. If you wish to allow use of your version of this file only
+ * under the terms of the GPL, and not to allow others to use your
+ * version of this file under the terms of the MPL, indicate your
+ * decision by deleting the provisions above and replace them with the
+ * notice and other provisions required by the GPL. If you do not
+ * delete the provisions above, a recipient may use your version of
+ * this file under the terms of any one of the MPL or the GPL.
+ *
+ * ***** END LICENSE BLOCK *****
+ */
+
+#include <stdio.h>
+#include <stdarg.h>
+#include <stdlib.h>
+
+#include "compat.h"
+
+int asprintf(char **strp, const char *fmt, ...)
+{
+	va_list ap;
+	int len;
+
+	va_start(ap, fmt);
+	len = _vscprintf(fmt, ap);
+	*strp = malloc(len+1);
+	if (!*strp)
+		return -1;
+
+	len = vsprintf(*strp, fmt, ap);
+	*strp[len] = 0;
+
+	va_end(ap);
+	return len;
+}
diff --git a/tools/windows/compat.h b/tools/windows/compat.h
new file mode 100644
index 0000000..8211b37
--- /dev/null
+++ b/tools/windows/compat.h
@@ -0,0 +1,51 @@
+/*
+ * ***** BEGIN LICENSE BLOCK *****
+ * Version: MPL 1.1/GPL 2.0
+ *
+ * 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 librabbitmq.
+ *
+ * The Initial Developers of the Original Code are LShift Ltd, Cohesive
+ * Financial Technologies LLC, and Rabbit Technologies Ltd.  Portions
+ * created before 22-Nov-2008 00:00:00 GMT by LShift Ltd, Cohesive
+ * Financial Technologies LLC, or Rabbit Technologies Ltd are Copyright
+ * (C) 2007-2008 LShift Ltd, Cohesive Financial Technologies LLC, and
+ * Rabbit Technologies Ltd.
+ *
+ * Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
+ * Ltd. Portions created by Cohesive Financial Technologies LLC are
+ * Copyright (C) 2007-2010 Cohesive Financial Technologies
+ * LLC. Portions created by Rabbit Technologies Ltd are Copyright (C)
+ * 2007-2010 Rabbit Technologies Ltd.
+ *
+ * Portions created by Tony Garnock-Jones are Copyright (C) 2009-2010
+ * LShift Ltd and Tony Garnock-Jones.
+ *
+ * All Rights Reserved.
+ *
+ * Contributor(s): ______________________________________.
+ *
+ * Alternatively, the contents of this file may be used under the terms
+ * of the GNU General Public License Version 2 or later (the "GPL"), in
+ * which case the provisions of the GPL are applicable instead of those
+ * above. If you wish to allow use of your version of this file only
+ * under the terms of the GPL, and not to allow others to use your
+ * version of this file under the terms of the MPL, indicate your
+ * decision by deleting the provisions above and replace them with the
+ * notice and other provisions required by the GPL. If you do not
+ * delete the provisions above, a recipient may use your version of
+ * this file under the terms of any one of the MPL or the GPL.
+ *
+ * ***** END LICENSE BLOCK *****
+ */
+
+extern int asprintf(char **strp, const char *fmt, ...);
diff --git a/tools/windows/process.c b/tools/windows/process.c
new file mode 100644
index 0000000..9a0b893
--- /dev/null
+++ b/tools/windows/process.c
@@ -0,0 +1,204 @@
+/*
+ * ***** BEGIN LICENSE BLOCK *****
+ * Version: MPL 1.1/GPL 2.0
+ *
+ * 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 librabbitmq.
+ *
+ * The Initial Developers of the Original Code are LShift Ltd, Cohesive
+ * Financial Technologies LLC, and Rabbit Technologies Ltd.  Portions
+ * created before 22-Nov-2008 00:00:00 GMT by LShift Ltd, Cohesive
+ * Financial Technologies LLC, or Rabbit Technologies Ltd are Copyright
+ * (C) 2007-2008 LShift Ltd, Cohesive Financial Technologies LLC, and
+ * Rabbit Technologies Ltd.
+ *
+ * Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
+ * Ltd. Portions created by Cohesive Financial Technologies LLC are
+ * Copyright (C) 2007-2010 Cohesive Financial Technologies
+ * LLC. Portions created by Rabbit Technologies Ltd are Copyright (C)
+ * 2007-2010 Rabbit Technologies Ltd.
+ *
+ * Portions created by Tony Garnock-Jones are Copyright (C) 2009-2010
+ * LShift Ltd and Tony Garnock-Jones.
+ *
+ * All Rights Reserved.
+ *
+ * Contributor(s): ______________________________________.
+ *
+ * Alternatively, the contents of this file may be used under the terms
+ * of the GNU General Public License Version 2 or later (the "GPL"), in
+ * which case the provisions of the GPL are applicable instead of those
+ * above. If you wish to allow use of your version of this file only
+ * under the terms of the GPL, and not to allow others to use your
+ * version of this file under the terms of the MPL, indicate your
+ * decision by deleting the provisions above and replace them with the
+ * notice and other provisions required by the GPL. If you do not
+ * delete the provisions above, a recipient may use your version of
+ * this file under the terms of any one of the MPL or the GPL.
+ *
+ * ***** END LICENSE BLOCK *****
+ */
+
+#include <stdio.h>
+#include <io.h>
+#include <windows.h>
+
+#include "common.h"
+#include "process.h"
+
+void die_windows_error(const char *fmt, ...)
+{
+	char *msg;
+
+	va_list ap;
+	va_start(ap, fmt);
+	vfprintf(stderr, fmt, ap);
+	va_end(ap);
+
+	if (!FormatMessage(FORMAT_MESSAGE_FROM_SYSTEM
+			       | FORMAT_MESSAGE_ALLOCATE_BUFFER,
+			   NULL, GetLastError(), 
+			   MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
+			   (LPSTR)&msg, 0, NULL))
+		msg = "(failed to retrieving Windows error message)";
+
+	fprintf(stderr, ": %s\n", msg);
+	exit(1);
+}
+
+static char *make_command_line(const char *const *argv)
+{
+	int i;
+	size_t len = 1; /* initial quotes */
+	char *buf;
+	char *dest;
+
+	/* calculate the length of the required buffer, making worst
+	   case assumptions for simplicity */
+	for (i = 0;;) {
+		len += strlen(argv[i]) * 2;
+
+		if (!argv[++i])
+			break;
+
+		len += 3; /* quotes, space, quotes */
+	}
+
+	len += 2; /* final quotes and the terminating zero */
+
+	dest = buf = malloc(len);
+	if (!buf)
+		die("allocating memory for subprocess command line");
+	
+	*dest++ = '\"';
+
+	for (i = 0;;) {
+		const char *src = argv[i];
+		for (;;) {
+			switch (*src) {
+			case 0:
+				goto done;
+
+			case '\"':
+			case '\\':
+				*dest++ = '\\';
+				/* fall through */
+
+			default:
+				*dest++ = *src++;
+				break;
+			}
+		}
+		done:
+
+		if (!argv[++i])
+			break;
+
+		*dest++ = '\"';
+		*dest++ = ' ';
+		*dest++ = '\"';
+	}
+
+	*dest++ = '\"';
+	*dest++ = 0;
+	return buf;
+}
+
+void pipeline(const char *const *argv, struct pipeline *pl)
+{
+	HANDLE in_read_handle, in_write_handle;
+	SECURITY_ATTRIBUTES sec_attr;
+	PROCESS_INFORMATION proc_info; 
+	STARTUPINFO start_info;
+	char *cmdline = make_command_line(argv);
+	
+	sec_attr.nLength = sizeof sec_attr;
+	sec_attr.bInheritHandle = TRUE;
+	sec_attr.lpSecurityDescriptor = NULL;
+
+	if (!CreatePipe(&in_read_handle, &in_write_handle, &sec_attr, 0))
+		die_windows_error("CreatePipe");
+
+	if (!SetHandleInformation(in_write_handle, HANDLE_FLAG_INHERIT, 0))
+		die_windows_error("SetHandleInformation");
+	
+	/* when in Rome... */
+	ZeroMemory(&proc_info, sizeof proc_info);
+	ZeroMemory(&start_info, sizeof start_info);
+	
+	start_info.cb = sizeof start_info;
+	start_info.dwFlags |= STARTF_USESTDHANDLES;
+
+	if ((start_info.hStdError = GetStdHandle(STD_ERROR_HANDLE))
+	                                             == INVALID_HANDLE_VALUE
+	    || (start_info.hStdOutput = GetStdHandle(STD_OUTPUT_HANDLE))
+	                                             == INVALID_HANDLE_VALUE)
+		die_windows_error("GetStdHandle");
+	
+	start_info.hStdInput = in_read_handle;
+
+	if (!CreateProcess(NULL, cmdline, NULL, NULL, TRUE, 0,
+			   NULL, NULL, &start_info, &proc_info))
+		die_windows_error("CreateProcess");
+
+	if (!CloseHandle(proc_info.hThread))
+		die_windows_error("CloseHandle for thread");
+	if (!CloseHandle(in_read_handle))
+		die_windows_error("CloseHandle");
+
+	pl->proc_handle = proc_info.hProcess;
+	pl->infd = _open_osfhandle((intptr_t)in_write_handle, 0);
+}
+
+int finish_pipeline(struct pipeline *pl)
+{
+	DWORD code;
+
+	if (close(pl->infd))
+		die_errno(errno, "close");
+	
+	for (;;) {
+		if (!GetExitCodeProcess(pl->proc_handle, &code))
+			die_windows_error("GetExitCodeProcess");
+		if (code != STILL_ACTIVE)
+			break;
+
+		if (WaitForSingleObject(pl->proc_handle, INFINITE)
+		                                           == WAIT_FAILED)
+			die_windows_error("WaitForSingleObject");
+	}
+
+	if (!CloseHandle(pl->proc_handle))
+		die_windows_error("CloseHandle for process");
+	
+	return code;
+}
diff --git a/tools/windows/process.h b/tools/windows/process.h
new file mode 100644
index 0000000..df276a7
--- /dev/null
+++ b/tools/windows/process.h
@@ -0,0 +1,59 @@
+/*
+ * ***** BEGIN LICENSE BLOCK *****
+ * Version: MPL 1.1/GPL 2.0
+ *
+ * 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 librabbitmq.
+ *
+ * The Initial Developers of the Original Code are LShift Ltd, Cohesive
+ * Financial Technologies LLC, and Rabbit Technologies Ltd.  Portions
+ * created before 22-Nov-2008 00:00:00 GMT by LShift Ltd, Cohesive
+ * Financial Technologies LLC, or Rabbit Technologies Ltd are Copyright
+ * (C) 2007-2008 LShift Ltd, Cohesive Financial Technologies LLC, and
+ * Rabbit Technologies Ltd.
+ *
+ * Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
+ * Ltd. Portions created by Cohesive Financial Technologies LLC are
+ * Copyright (C) 2007-2010 Cohesive Financial Technologies
+ * LLC. Portions created by Rabbit Technologies Ltd are Copyright (C)
+ * 2007-2010 Rabbit Technologies Ltd.
+ *
+ * Portions created by Tony Garnock-Jones are Copyright (C) 2009-2010
+ * LShift Ltd and Tony Garnock-Jones.
+ *
+ * All Rights Reserved.
+ *
+ * Contributor(s): ______________________________________.
+ *
+ * Alternatively, the contents of this file may be used under the terms
+ * of the GNU General Public License Version 2 or later (the "GPL"), in
+ * which case the provisions of the GPL are applicable instead of those
+ * above. If you wish to allow use of your version of this file only
+ * under the terms of the GPL, and not to allow others to use your
+ * version of this file under the terms of the MPL, indicate your
+ * decision by deleting the provisions above and replace them with the
+ * notice and other provisions required by the GPL. If you do not
+ * delete the provisions above, a recipient may use your version of
+ * this file under the terms of any one of the MPL or the GPL.
+ *
+ * ***** END LICENSE BLOCK *****
+ */
+
+#include <windef.h>
+
+struct pipeline {
+	HANDLE proc_handle;
+	int infd;
+};
+
+extern void pipeline(const char *const *argv, struct pipeline *pl);
+extern int finish_pipeline(struct pipeline *pl);