libwinpr-utils: initial MessagePipe implementation

winpr/include/winpr/collections.h

@@ -314,4 +314,16 @@ WINPR_API int MessageQueue_Peek(wMessageQueue* queue, wMessage* message, BOOL re
 WINPR_API wMessageQueue* MessageQueue_New(void);
 WINPR_API void MessageQueue_Free(wMessageQueue* queue);
 
+/* Message Pipe */
+
+struct _wMessagePipe
+{
+	wMessageQueue* In;
+	wMessageQueue* Out;
+};
+typedef struct _wMessagePipe wMessagePipe;
+
+WINPR_API wMessagePipe* MessagePipe_New(void);
+WINPR_API void MessagePipe_Free(wMessagePipe* pipe);
+
 #endif /* WINPR_COLLECTIONS_H */

winpr/libwinpr/utils/CMakeLists.txt

@@ -29,7 +29,8 @@ set(${MODULE_PREFIX}_COLLECTIONS_SRCS
 	collections/CountdownEvent.c
 	collections/BufferPool.c
 	collections/ObjectPool.c
-	collections/MessageQueue.c)
+	collections/MessageQueue.c
+	collections/MessagePipe.c)
 
 set(${MODULE_PREFIX}_SRCS
 	sam.c

winpr/libwinpr/utils/collections/MessagePipe.c

@@ -0,0 +1,73 @@
+/**
+ * WinPR: Windows Portable Runtime
+ * Message Pipe
+ *
+ * Copyright 2012 Marc-Andre Moreau <marcandre.moreau@gmail.com>
+ *
+ * Licensed under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include <winpr/crt.h>
+#include <winpr/sysinfo.h>
+
+#include <winpr/collections.h>
+
+/**
+ * Properties
+ */
+
+
+
+/**
+ * Methods
+ */
+
+void MessagePipe_PostQuit(wMessagePipe* pipe, int nExitCode)
+{
+	MessageQueue_PostQuit(pipe->In, nExitCode);
+	MessageQueue_PostQuit(pipe->Out, nExitCode);
+}
+
+/**
+ * Construction, Destruction
+ */
+
+wMessagePipe* MessagePipe_New()
+{
+	wMessagePipe* pipe = NULL;
+
+	pipe = (wMessagePipe*) malloc(sizeof(wMessagePipe));
+
+	if (pipe)
+	{
+		pipe->In = MessageQueue_New();
+		pipe->Out = MessageQueue_New();
+	}
+
+	return pipe;
+}
+
+void MessagePipe_Free(wMessagePipe* pipe)
+{
+	if (pipe)
+	{
+		MessageQueue_Free(pipe->In);
+		MessageQueue_Free(pipe->Out);
+
+		free(pipe);
+	}
+}

winpr/libwinpr/utils/test/CMakeLists.txt

@@ -8,7 +8,8 @@ set(${MODULE_PREFIX}_TESTS
 	TestQueue.c
 	TestArrayList.c
 	TestCmdLine.c
-	TestMessageQueue.c)
+	TestMessageQueue.c
+	TestMessagePipe.c)
 
 create_test_sourcelist(${MODULE_PREFIX}_SRCS
 	${${MODULE_PREFIX}_DRIVER}

winpr/libwinpr/utils/test/TestMessagePipe.c

@@ -0,0 +1,81 @@
+
+#include <winpr/crt.h>
+#include <winpr/thread.h>
+#include <winpr/collections.h>
+
+static void* message_echo_pipe_client_thread(void* arg)
+{
+	int index;
+	int count;
+	wMessage message;
+	wMessagePipe* pipe;
+
+	count = index = 0;
+	pipe = (wMessagePipe*) arg;
+
+	while (index < 100)
+	{
+		MessageQueue_Post(pipe->In, NULL, 0, (void*) (size_t) index, NULL);
+
+		if (!MessageQueue_Wait(pipe->Out))
+			break;
+
+		if (!MessageQueue_Peek(pipe->Out, &message, TRUE))
+			break;
+
+		if (message.id == WMQ_QUIT)
+			break;
+
+		count = (int) (size_t) message.wParam;
+
+		if (count != index)
+			printf("Echo count mismatch: Actual: %d, Expected: %d\n", count, index);
+
+		index++;
+	}
+
+	MessageQueue_PostQuit(pipe->In, 0);
+
+	return NULL;
+}
+
+static void* message_echo_pipe_server_thread(void* arg)
+{
+	int count;
+	wMessage message;
+	wMessagePipe* pipe;
+
+	pipe = (wMessagePipe*) arg;
+
+	while (MessageQueue_Wait(pipe->In))
+	{
+		if (MessageQueue_Peek(pipe->In, &message, TRUE))
+		{
+			if (message.id == WMQ_QUIT)
+				break;
+
+			count = (int) (size_t) message.wParam;
+
+			MessageQueue_Dispatch(pipe->Out, &message);
+		}
+	}
+
+	return NULL;
+}
+
+int TestMessagePipe(int argc, char* argv[])
+{
+	HANDLE ClientThread;
+	HANDLE ServerThread;
+	wMessagePipe* EchoPipe;
+
+	EchoPipe = MessagePipe_New();
+
+	ClientThread = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE) message_echo_pipe_client_thread, (void*) EchoPipe, 0, NULL);
+	ServerThread = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE) message_echo_pipe_server_thread, (void*) EchoPipe, 0, NULL);
+
+	WaitForSingleObject(ClientThread, INFINITE);
+	WaitForSingleObject(ServerThread, INFINITE);
+
+	return 0;
+}