mqueue.c

内容正如其名

/******************************************************************************
  Copyright(C) 2005,2006 Frank ZHANG

  All Rights Reserved.
    
  This program is free software; you can redistribute it and/or modify it
  under the terms of the GNU General Public License as published by the Free
  Software Foundation; either version 2 of the License, or (at your option)
  any later version.

  This program is distributed in the hope that it will be useful, but WITHOUT
  ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or 
  FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for 
  more details.
    
  You should have received a copy of the GNU General Public License along with
  this program; if not, write to the Free Software Foundation, Inc., 59 Temple
  Place, Suite 330, Boston, MA 02111-1307 USA.
 
 ******************************************************************************
 
  If you want to distribute this software with modifications under any terms
  other than the GPL or distribute the software linked with proprietary
  applications that are not distributed in full compliance with the GNU Public
  License, a Commercial License is needed.
  
  Commercial licensing and support of this software are available at a fee. For
  more information, See http://www.openmgcp.org
 
 ******************************************************************************/

#include <windows.h>
#include <stdio.h>

#include "assert.h"
#include "list.h"
#include "mqueue.h"

static SLIST MsgqList;
static HANDLE hMutex;

typedef struct MSQ_tag
{
  HANDLE hReadPipe;
  HANDLE hWritePipe;

  struct mq_attr Attri;
  char Name[100];
} MSQ;


int mq_close(mqd_t mqdes)
{
  MSQ *pMsq = (MSQ*)mqdes;

  WaitForSingleObject(hMutex, INFINITE);
  SListDelNodeByData(&MsgqList, pMsq);
  ReleaseMutex(hMutex);

  CloseHandle(pMsq->hReadPipe);
  CloseHandle(pMsq->hWritePipe);
  free(pMsq);

  if (MsgqList.count <= 0)
  {
    CloseHandle(hMutex);
    hMutex = NULL;
  }

  return 0;
}

mqd_t mq_open(const char *name, int oflag, ...)
{
  MSQ *pMsq = NULL;
  va_list vaCreateArgs;
  struct mq_attr *pAttr;
  
  SListReset(&MsgqList);
  while ((pMsq = SListGetCurData(&MsgqList)) != NULL)
  {
    if (strcmp(pMsq->Name, name) == 0)
      return (mqd_t)pMsq;

    SListNextNode(&MsgqList);
  }

  pMsq = calloc(1, sizeof(MSQ));
  strcpy(pMsq->Name, name);

	va_start(vaCreateArgs, oflag);
  
	/* get the mode argument, ignore it */
	va_arg(vaCreateArgs, int);
  
	/* get a pointer to an mq_attr structure */
	pAttr = va_arg(vaCreateArgs, struct mq_attr*);
	va_end(vaCreateArgs);
 	memcpy(&pMsq->Attri, pAttr, sizeof(struct mq_attr));

  if (!CreatePipe(&pMsq->hReadPipe, &pMsq->hWritePipe, NULL,
                  pMsq->Attri.mq_msgsize*pMsq->Attri.mq_maxmsg))
  {
    printf("\nFail to create Pipe! %d", GetLastError());
    return -1;
  }

  if (hMutex == NULL)
    hMutex = CreateMutex(NULL,FALSE, "MutexToProtectMsq");

  assert(hMutex);

  
  WaitForSingleObject(hMutex, INFINITE);
  SListAppend(&MsgqList, pMsq);
  ReleaseMutex(hMutex);

  return (mqd_t)pMsq;
}

int mq_receive(mqd_t mqdes, char *msg_ptr, size_t msg_len, unsigned *msg_prio)
{
  MSQ *pMsq = (MSQ *)mqdes;
  DWORD lpNumberOfBytesRead = 0;
  assert(mqdes);

  /* Check message size validity */
  if (msg_len != (unsigned int)pMsq->Attri.mq_msgsize)
  {
    assert(0);
    return -1;
  }

  if (ReadFile(pMsq->hReadPipe, msg_ptr, msg_len, &lpNumberOfBytesRead, NULL))
    return msg_len;
  
  printf("\nFail to Receive msg!");

  return -1;
}

int mq_send(mqd_t mqdes, const char *msg_ptr, size_t msg_len, unsigned msg_prio)
{
  MSQ *pMsq = (MSQ *)mqdes;
  DWORD lpNumberOfBytesWritten = 0;
  assert(mqdes);

  /* Check message size validity */
  if (msg_len != (unsigned int)pMsq->Attri.mq_msgsize)
  {
    assert(0);
    return -1;
  }

  if (WriteFile(pMsq->hWritePipe, msg_ptr, msg_len, &lpNumberOfBytesWritten, NULL))
	  return 0;
  
  printf("\nFail to send msg!");
  
  return -1;
}

int mq_timedreceive(mqd_t mqdes, char *msg_ptr, size_t msg_len,
                          unsigned *msg_prio, const struct timespec *abs_timeout)
{
  return -1;
}

int mq_timedsend(mqd_t mqdes, const char *msg_ptr, size_t msg_len,
                      unsigned msg_prio, const struct timespec *abs_timeout)
{
  return mq_send(mqdes, msg_ptr, msg_len, msg_prio);
}

int mq_unlink(const char *name) 
{
  //printf("Call mq_unlink!\n");
  return 0;
}