Blame winpr/libwinpr/utils/collections/MessageQueue.c

Packit 1fb8d4
/**
Packit 1fb8d4
 * WinPR: Windows Portable Runtime
Packit 1fb8d4
 * Message Queue
Packit 1fb8d4
 *
Packit 1fb8d4
 * Copyright 2012 Marc-Andre Moreau <marcandre.moreau@gmail.com>
Packit 1fb8d4
 *
Packit 1fb8d4
 * Licensed under the Apache License, Version 2.0 (the "License");
Packit 1fb8d4
 * you may not use this file except in compliance with the License.
Packit 1fb8d4
 * You may obtain a copy of the License at
Packit 1fb8d4
 *
Packit 1fb8d4
 *     http://www.apache.org/licenses/LICENSE-2.0
Packit 1fb8d4
 *
Packit 1fb8d4
 * Unless required by applicable law or agreed to in writing, software
Packit 1fb8d4
 * distributed under the License is distributed on an "AS IS" BASIS,
Packit 1fb8d4
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
Packit 1fb8d4
 * See the License for the specific language governing permissions and
Packit 1fb8d4
 * limitations under the License.
Packit 1fb8d4
 */
Packit 1fb8d4
Packit 1fb8d4
#ifdef HAVE_CONFIG_H
Packit 1fb8d4
#include "config.h"
Packit 1fb8d4
#endif
Packit 1fb8d4
Packit 1fb8d4
#include <winpr/crt.h>
Packit 1fb8d4
#include <winpr/sysinfo.h>
Packit 1fb8d4
Packit 1fb8d4
#include <winpr/collections.h>
Packit 1fb8d4
Packit 1fb8d4
/**
Packit 1fb8d4
 * Message Queue inspired from Windows:
Packit 1fb8d4
 * http://msdn.microsoft.com/en-us/library/ms632590/
Packit 1fb8d4
 */
Packit 1fb8d4
Packit 1fb8d4
/**
Packit 1fb8d4
 * Properties
Packit 1fb8d4
 */
Packit 1fb8d4
Packit 1fb8d4
/**
Packit 1fb8d4
 * Gets an event which is set when the queue is non-empty
Packit 1fb8d4
 */
Packit 1fb8d4
Packit 1fb8d4
HANDLE MessageQueue_Event(wMessageQueue* queue)
Packit 1fb8d4
{
Packit 1fb8d4
	return queue->event;
Packit 1fb8d4
}
Packit 1fb8d4
Packit 1fb8d4
/**
Packit 1fb8d4
 * Gets the queue size
Packit 1fb8d4
 */
Packit 1fb8d4
Packit 1fb8d4
int MessageQueue_Size(wMessageQueue* queue)
Packit 1fb8d4
{
Packit 1fb8d4
	return queue->size;
Packit 1fb8d4
}
Packit 1fb8d4
Packit 1fb8d4
/**
Packit 1fb8d4
 * Methods
Packit 1fb8d4
 */
Packit 1fb8d4
Packit 1fb8d4
BOOL MessageQueue_Wait(wMessageQueue* queue)
Packit 1fb8d4
{
Packit 1fb8d4
	BOOL status = FALSE;
Packit 1fb8d4
Packit 1fb8d4
	if (WaitForSingleObject(queue->event, INFINITE) == WAIT_OBJECT_0)
Packit 1fb8d4
		status = TRUE;
Packit 1fb8d4
Packit 1fb8d4
	return status;
Packit 1fb8d4
}
Packit 1fb8d4
Packit 1fb8d4
BOOL MessageQueue_Dispatch(wMessageQueue* queue, wMessage* message)
Packit 1fb8d4
{
Packit 1fb8d4
	BOOL ret = FALSE;
Packit Service 5a9772
	if (!queue || !message)
Packit Service 5a9772
		return FALSE;
Packit Service 5a9772
Packit 1fb8d4
	EnterCriticalSection(&queue->lock);
Packit 1fb8d4
Packit 1fb8d4
	if (queue->size == queue->capacity)
Packit 1fb8d4
	{
Packit 1fb8d4
		int old_capacity;
Packit 1fb8d4
		int new_capacity;
Packit 1fb8d4
		wMessage* new_arr;
Packit 1fb8d4
Packit 1fb8d4
		old_capacity = queue->capacity;
Packit 1fb8d4
		new_capacity = queue->capacity * 2;
Packit 1fb8d4
Packit Service 5a9772
		new_arr = (wMessage*)realloc(queue->array, sizeof(wMessage) * new_capacity);
Packit 1fb8d4
		if (!new_arr)
Packit 1fb8d4
			goto out;
Packit 1fb8d4
		queue->array = new_arr;
Packit 1fb8d4
		queue->capacity = new_capacity;
Packit 1fb8d4
		ZeroMemory(&(queue->array[old_capacity]), (new_capacity - old_capacity) * sizeof(wMessage));
Packit 1fb8d4
Packit 1fb8d4
		/* rearrange wrapped entries */
Packit 1fb8d4
		if (queue->tail <= queue->head)
Packit 1fb8d4
		{
Packit 1fb8d4
			CopyMemory(&(queue->array[old_capacity]), queue->array, queue->tail * sizeof(wMessage));
Packit 1fb8d4
			queue->tail += old_capacity;
Packit 1fb8d4
		}
Packit 1fb8d4
	}
Packit 1fb8d4
Packit 1fb8d4
	CopyMemory(&(queue->array[queue->tail]), message, sizeof(wMessage));
Packit 1fb8d4
	queue->tail = (queue->tail + 1) % queue->capacity;
Packit 1fb8d4
	queue->size++;
Packit 1fb8d4
Packit 1fb8d4
	message = &(queue->array[queue->tail]);
Packit Service 5a9772
	message->time = GetTickCount64();
Packit 1fb8d4
Packit 1fb8d4
	if (queue->size > 0)
Packit 1fb8d4
		SetEvent(queue->event);
Packit 1fb8d4
Packit 1fb8d4
	ret = TRUE;
Packit 1fb8d4
out:
Packit 1fb8d4
	LeaveCriticalSection(&queue->lock);
Packit 1fb8d4
	return ret;
Packit 1fb8d4
}
Packit 1fb8d4
Packit 1fb8d4
BOOL MessageQueue_Post(wMessageQueue* queue, void* context, UINT32 type, void* wParam, void* lParam)
Packit 1fb8d4
{
Packit 1fb8d4
	wMessage message;
Packit 1fb8d4
Packit 1fb8d4
	message.context = context;
Packit 1fb8d4
	message.id = type;
Packit 1fb8d4
	message.wParam = wParam;
Packit 1fb8d4
	message.lParam = lParam;
Packit 1fb8d4
	message.Free = NULL;
Packit 1fb8d4
Packit 1fb8d4
	return MessageQueue_Dispatch(queue, &message);
Packit 1fb8d4
}
Packit 1fb8d4
Packit 1fb8d4
BOOL MessageQueue_PostQuit(wMessageQueue* queue, int nExitCode)
Packit 1fb8d4
{
Packit Service 5a9772
	return MessageQueue_Post(queue, NULL, WMQ_QUIT, (void*)(size_t)nExitCode, NULL);
Packit 1fb8d4
}
Packit 1fb8d4
Packit 1fb8d4
int MessageQueue_Get(wMessageQueue* queue, wMessage* message)
Packit 1fb8d4
{
Packit 1fb8d4
	int status = -1;
Packit 1fb8d4
Packit 1fb8d4
	if (!MessageQueue_Wait(queue))
Packit 1fb8d4
		return status;
Packit 1fb8d4
Packit 1fb8d4
	EnterCriticalSection(&queue->lock);
Packit 1fb8d4
Packit 1fb8d4
	if (queue->size > 0)
Packit 1fb8d4
	{
Packit 1fb8d4
		CopyMemory(message, &(queue->array[queue->head]), sizeof(wMessage));
Packit 1fb8d4
		ZeroMemory(&(queue->array[queue->head]), sizeof(wMessage));
Packit 1fb8d4
		queue->head = (queue->head + 1) % queue->capacity;
Packit 1fb8d4
		queue->size--;
Packit 1fb8d4
Packit 1fb8d4
		if (queue->size < 1)
Packit 1fb8d4
			ResetEvent(queue->event);
Packit 1fb8d4
Packit 1fb8d4
		status = (message->id != WMQ_QUIT) ? 1 : 0;
Packit 1fb8d4
	}
Packit 1fb8d4
Packit 1fb8d4
	LeaveCriticalSection(&queue->lock);
Packit 1fb8d4
Packit 1fb8d4
	return status;
Packit 1fb8d4
}
Packit 1fb8d4
Packit 1fb8d4
int MessageQueue_Peek(wMessageQueue* queue, wMessage* message, BOOL remove)
Packit 1fb8d4
{
Packit 1fb8d4
	int status = 0;
Packit 1fb8d4
Packit 1fb8d4
	EnterCriticalSection(&queue->lock);
Packit 1fb8d4
Packit 1fb8d4
	if (queue->size > 0)
Packit 1fb8d4
	{
Packit 1fb8d4
		CopyMemory(message, &(queue->array[queue->head]), sizeof(wMessage));
Packit 1fb8d4
		status = 1;
Packit 1fb8d4
Packit 1fb8d4
		if (remove)
Packit 1fb8d4
		{
Packit 1fb8d4
			ZeroMemory(&(queue->array[queue->head]), sizeof(wMessage));
Packit 1fb8d4
			queue->head = (queue->head + 1) % queue->capacity;
Packit 1fb8d4
			queue->size--;
Packit 1fb8d4
Packit 1fb8d4
			if (queue->size < 1)
Packit 1fb8d4
				ResetEvent(queue->event);
Packit 1fb8d4
		}
Packit 1fb8d4
	}
Packit 1fb8d4
Packit 1fb8d4
	LeaveCriticalSection(&queue->lock);
Packit 1fb8d4
Packit 1fb8d4
	return status;
Packit 1fb8d4
}
Packit 1fb8d4
Packit 1fb8d4
/**
Packit 1fb8d4
 * Construction, Destruction
Packit 1fb8d4
 */
Packit 1fb8d4
Packit Service 5a9772
wMessageQueue* MessageQueue_New(const wObject* callback)
Packit 1fb8d4
{
Packit 1fb8d4
	wMessageQueue* queue = NULL;
Packit 1fb8d4
Packit Service 5a9772
	queue = (wMessageQueue*)calloc(1, sizeof(wMessageQueue));
Packit 1fb8d4
	if (!queue)
Packit 1fb8d4
		return NULL;
Packit 1fb8d4
Packit 1fb8d4
	queue->capacity = 32;
Packit Service 5a9772
	queue->array = (wMessage*)calloc(queue->capacity, sizeof(wMessage));
Packit 1fb8d4
	if (!queue->array)
Packit 1fb8d4
		goto error_array;
Packit 1fb8d4
Packit 1fb8d4
	if (!InitializeCriticalSectionAndSpinCount(&queue->lock, 4000))
Packit 1fb8d4
		goto error_spinlock;
Packit 1fb8d4
Packit 1fb8d4
	queue->event = CreateEvent(NULL, TRUE, FALSE, NULL);
Packit 1fb8d4
	if (!queue->event)
Packit 1fb8d4
		goto error_event;
Packit 1fb8d4
Packit 1fb8d4
	if (callback)
Packit 1fb8d4
		queue->object = *callback;
Packit 1fb8d4
Packit 1fb8d4
	return queue;
Packit 1fb8d4
Packit 1fb8d4
error_event:
Packit 1fb8d4
	DeleteCriticalSection(&queue->lock);
Packit 1fb8d4
error_spinlock:
Packit 1fb8d4
	free(queue->array);
Packit 1fb8d4
error_array:
Packit 1fb8d4
	free(queue);
Packit 1fb8d4
	return NULL;
Packit 1fb8d4
}
Packit 1fb8d4
Packit 1fb8d4
void MessageQueue_Free(wMessageQueue* queue)
Packit 1fb8d4
{
Packit 1fb8d4
	if (!queue)
Packit 1fb8d4
		return;
Packit 1fb8d4
Packit 1fb8d4
	MessageQueue_Clear(queue);
Packit 1fb8d4
Packit 1fb8d4
	CloseHandle(queue->event);
Packit 1fb8d4
	DeleteCriticalSection(&queue->lock);
Packit 1fb8d4
Packit 1fb8d4
	free(queue->array);
Packit 1fb8d4
	free(queue);
Packit 1fb8d4
}
Packit 1fb8d4
Packit Service 5a9772
int MessageQueue_Clear(wMessageQueue* queue)
Packit 1fb8d4
{
Packit 1fb8d4
	int status = 0;
Packit 1fb8d4
Packit 1fb8d4
	EnterCriticalSection(&queue->lock);
Packit 1fb8d4
Packit Service 5a9772
	while (queue->size > 0)
Packit 1fb8d4
	{
Packit Service 5a9772
		wMessage* msg = &(queue->array[queue->head]);
Packit 1fb8d4
Packit 1fb8d4
		/* Free resources of message. */
Packit 1fb8d4
		if (queue->object.fnObjectUninit)
Packit 1fb8d4
			queue->object.fnObjectUninit(msg);
Packit 1fb8d4
		if (queue->object.fnObjectFree)
Packit 1fb8d4
			queue->object.fnObjectFree(msg);
Packit 1fb8d4
Packit 1fb8d4
		ZeroMemory(msg, sizeof(wMessage));
Packit 1fb8d4
Packit 1fb8d4
		queue->head = (queue->head + 1) % queue->capacity;
Packit 1fb8d4
		queue->size--;
Packit 1fb8d4
	}
Packit 1fb8d4
	ResetEvent(queue->event);
Packit 1fb8d4
Packit 1fb8d4
	LeaveCriticalSection(&queue->lock);
Packit 1fb8d4
Packit 1fb8d4
	return status;
Packit 1fb8d4
}