struct _EMsgPort {
	GAsyncQueue *queue;
	EMsg *cache;
	gint pipe[2];  /* on Win32, actually a pair of SOCKETs */
#ifdef HAVE_NSS
	PRFileDesc *prpipe[2];
#endif
};

/* message flags */
enum {
	MSG_FLAG_SYNC_WITH_PIPE    = 1 << 0,
	MSG_FLAG_SYNC_WITH_PR_PIPE = 1 << 1
};

#ifdef HAVE_NSS
static int
e_prpipe (PRFileDesc **fds)
{
#ifdef G_OS_WIN32
	if (PR_NewTCPSocketPair (fds) != PR_FAILURE)
		return 0;
#else
	if (PR_CreatePipe (&fds[0], &fds[1]) != PR_FAILURE)
		return 0;
#endif
	
	fds[0] = NULL;
	fds[1] = NULL;
	
	return -1;
}
#endif

static void
msgport_sync_with_pipe (gint fd)
{
	gchar buffer[1];

	while (fd >= 0) {
		if (E_READ (fd, buffer, 1) > 0)
			break;
		else if (!E_IS_STATUS_INTR ()) {
			g_warning ("%s: Failed to read from pipe: %s",
				G_STRFUNC, g_strerror (errno));
			break;
		}
	}
}

#ifdef HAVE_NSS
static void
msgport_sync_with_prpipe (PRFileDesc *prfd)
{
	gchar buffer[1];

	while (prfd != NULL) {
		if (PR_Read (prfd, buffer, 1) > 0)
			break;
		else if (PR_GetError () != PR_PENDING_INTERRUPT_ERROR) {
			gchar *text = g_alloca (PR_GetErrorTextLength ());
			PR_GetErrorText (text);
			g_warning ("%s: Failed to read from NSPR pipe: %s",
				G_STRFUNC, text);
			break;
		}
	}
}
#endif

EMsgPort *
e_msgport_new (void)
{
	EMsgPort *msgport;

	msgport = g_slice_new (EMsgPort);
	msgport->queue = g_async_queue_new ();
	msgport->cache = NULL;
	msgport->pipe[0] = -1;
	msgport->pipe[1] = -1;
#ifdef HAVE_NSS
	msgport->prpipe[0] = NULL;
	msgport->prpipe[1] = NULL;
#endif

	return msgport;
}

void
e_msgport_destroy (EMsgPort *msgport)
{
	g_return_if_fail (msgport != NULL);

	if (msgport->pipe[0] >= 0) {
		E_CLOSE (msgport->pipe[0]);
		E_CLOSE (msgport->pipe[1]);
	}
#ifdef HAVE_NSS
	if (msgport->prpipe[0] != NULL) {
		PR_Close (msgport->prpipe[0]);
		PR_Close (msgport->prpipe[1]);
	}
#endif

	g_async_queue_unref (msgport->queue);
	g_slice_free (EMsgPort, msgport);
}

int
e_msgport_fd (EMsgPort *msgport)
{
	gint fd;

	g_return_val_if_fail (msgport != NULL, -1);

	g_async_queue_lock (msgport->queue);
	fd = msgport->pipe[0];
	if (fd < 0 && e_pipe (msgport->pipe) == 0)
		fd = msgport->pipe[0];
	g_async_queue_unlock (msgport->queue);

	return fd;
}

#ifdef HAVE_NSS
PRFileDesc *
e_msgport_prfd (EMsgPort *msgport)
{
	PRFileDesc *prfd;

	g_return_val_if_fail (msgport != NULL, NULL);

	g_async_queue_lock (msgport->queue);
	prfd = msgport->prpipe[0];
	if (prfd == NULL && e_prpipe (msgport->prpipe) == 0)
		prfd = msgport->prpipe[0];
	g_async_queue_unlock (msgport->queue);

	return prfd;
}
#endif

void
e_msgport_put (EMsgPort *msgport, EMsg *msg)
{
	gint fd;
#ifdef HAVE_NSS
	PRFileDesc *prfd;
#endif

	g_return_if_fail (msgport != NULL);
	g_return_if_fail (msg != NULL);

	g_async_queue_lock (msgport->queue);

	msg->flags = 0;

	fd = msgport->pipe[1];
	while (fd >= 0) {
		if (E_WRITE (fd, "E", 1) > 0) {
			msg->flags |= MSG_FLAG_SYNC_WITH_PIPE;
			break;
		} else if (!E_IS_STATUS_INTR ()) {
			g_warning ("%s: Failed to write to pipe: %s",
				G_STRFUNC, g_strerror (errno));
			break;
		}
	}

#ifdef HAVE_NSS
	prfd = msgport->prpipe[1];
	while (prfd != NULL) {
		if (PR_Write (prfd, "E", 1) > 0) {
			msg->flags |= MSG_FLAG_SYNC_WITH_PR_PIPE;
			break;
		} else if (PR_GetError () != PR_PENDING_INTERRUPT_ERROR) {
			gchar *text = g_alloca (PR_GetErrorTextLength ());
			PR_GetErrorText (text);
			g_warning ("%s: Failed to write to NSPR pipe: %s",
				G_STRFUNC, text);
			break;
		}
	}
#endif

	g_async_queue_push_unlocked (msgport->queue, msg);
	g_async_queue_unlock (msgport->queue);
}

EMsg *
e_msgport_wait (EMsgPort *msgport)
{
	EMsg *msg;

	g_return_val_if_fail (msgport != NULL, NULL);

	g_async_queue_lock (msgport->queue);

	/* check the cache first */
	if (msgport->cache != NULL) {
		msg = msgport->cache;
		/* don't clear the cache */
		g_async_queue_unlock (msgport->queue);
		return msg;
	}

	msg = g_async_queue_pop_unlocked (msgport->queue);

	g_assert (msg != NULL);

	/* The message is not actually "removed" from the EMsgPort until
 	 * e_msgport_get() is called.  So we cache the popped message. */
	msgport->cache = msg;

	if (msg->flags & MSG_FLAG_SYNC_WITH_PIPE)
		msgport_sync_with_pipe (msgport->pipe[0]);
#ifdef HAVE_NSS
	if (msg->flags & MSG_FLAG_SYNC_WITH_PR_PIPE)
		msgport_sync_with_prpipe (msgport->prpipe[0]);
#endif

	g_async_queue_unlock (msgport->queue);

	return msg;
}

EMsg *
e_msgport_get (EMsgPort *msgport)
{
	EMsg *msg;

	g_return_val_if_fail (msgport != NULL, NULL);

	g_async_queue_lock (msgport->queue);

	/* check the cache first */
	if (msgport->cache != NULL) {
		msg = msgport->cache;
		msgport->cache = NULL;
		g_async_queue_unlock (msgport->queue);
		return msg;
	}

	msg = g_async_queue_try_pop_unlocked (msgport->queue);

	if (msg != NULL && msg->flags & MSG_FLAG_SYNC_WITH_PIPE)
		msgport_sync_with_pipe (msgport->pipe[0]);
#ifdef HAVE_NSS
	if (msg != NULL && msg->flags & MSG_FLAG_SYNC_WITH_PR_PIPE)
		msgport_sync_with_prpipe (msgport->prpipe[0]);
#endif

	g_async_queue_unlock (msgport->queue);

	return msg;
}

void
e_msgport_reply (EMsg *msg)
{
	g_return_if_fail (msg != NULL);

	if (msg->reply_port)
		e_msgport_put (msg->reply_port, msg);

	/* else lost? */
}
