myvtm/libtoolkit/ioqueue.h

#ifndef IOQUEUE_H
#define IOQUEUE_H

#pragma once

#include "config.h"
#include <winpr/winsock.h>

#ifdef __cplusplus
extern "C" {
#endif

#include <stddef.h>

#include "timerqueue.h"
#include "list.h"
#include "spinlock.h"

////////////////////////////////////////////////////////////////////////////////////////////////
// ioqueue
// 
// support: 
//		(1) user message queue
//		(2) timer
//		(3) proactor network framework

/**
 * opaque structure for io queue
 */
typedef struct ioqueue_t ioqueue_t;

/**
 * @return return non-zero means need continue processing
 */
typedef int (*ioqueue_on_msg_callback)(unsigned short msg_id, int param1, int param2);

#define MAX_MSG						1024
#define MAX_MSG_PRIORITY			32

/**
 * create io queue
 * @param ioq [out] pointer to ioqueue_t structure
 * @return NULL failed
 */
TOOLKIT_API ioqueue_t *ioqueue_create();

/**
 * destroy io queue
 */
TOOLKIT_API void ioqueue_destroy(ioqueue_t *ioq);

/**
 * is all handler closed ?
 */
TOOLKIT_API int ioqueue_handler_empty(ioqueue_t *ioq);

/** is all pending msg procceed ? */
TOOLKIT_API int ioqueue_msg_empty(ioqueue_t *ioq);

/** detect whether can exit */
static int __inline ioqueue_can_exit(ioqueue_t *ioq) 
{ 
	return ioqueue_handler_empty(ioq) && ioqueue_msg_empty(ioq);
}

/**
 * set user cutomized data
 * @return the old user_data value
 */
TOOLKIT_API void *ioqueue_set_user_data(ioqueue_t *ioq, void* user_data);

/**
 * get user cutomized data
 */
TOOLKIT_API void *ioqueue_get_user_data(ioqueue_t *ioq);

/**
 * poll io queue for dispatching events
 * @param timeout time out value for a poll operation
 * @return number of events dispatched
 */
TOOLKIT_API int ioqueue_poll(ioqueue_t *ioq, int timeout);

/**
 * stop
 */
TOOLKIT_API void ioqueue_stop(ioqueue_t *ioq);

/**
 * register a message handler for msg_type
 * @param msg_type message type identifier [0, 1024)
 * @param priority [0,32), the lower value means this cb will be invoked first while processing msgs of msg_id
 * @return -1 failed, 0 success
 */
TOOLKIT_API int ioqueue_msg_add_handler(ioqueue_t *ioq, int msg_type, int priority, ioqueue_on_msg_callback cb);

/** 
 * unregister a message handler for msg_type
 */
TOOLKIT_API int ioqueue_msg_remove_handler(ioqueue_t *ioq, int msg_type, int priority);

/** 
 * post user message
 */
TOOLKIT_API int ioqueue_post_message(ioqueue_t *ioq, int msg_type, int param1, int param2);

/**
 * send message, synchronous operation
 */
TOOLKIT_API int ioqueue_send_message(ioqueue_t *ioq, int msg_type, int param1, int param2);

/**
 * cancel all message with msg_type
 */
// int ioqueue_cancel_message(ioqueue_t *ioq, int msg_type);

/**
 * start a timer invoked at now() + delay
 * @param entry the entry to schedule
 * @param delay time value to delay, in milliseconds
 * @return 0 success, -1 failed
 */
TOOLKIT_API int ioqueue_timer_schedule(ioqueue_t *ioq, timer_entry *entry, unsigned int delay);

/**
 * cancel a timer 
 * @param entry timer entry that has been scheduled 
 * @param cancel if cancel successful if invoke timer with error = -1? 
 * @return 0 cancel ok, -1 failed
 */
TOOLKIT_API int ioqueue_timer_cancel(ioqueue_t *ioq, timer_entry *entry, int cancel);

////////////////////////////////////////////////////////////////////////////////////////////////
// acceptor

typedef struct ioqueue_handle_context { /* do not access directly */
	struct list_head node;
	int type;
	union {
		HANDLE file;
		SOCKET sock;
		char *pipe_name;
	}u;
	void *user_data;
	ioqueue_t *owner;
	LONG pending_ios;
	lock_t ov_pending_list_lock;
	struct list_head ov_pending_list;
}ioqueue_handle_context;

typedef struct ioqueue_handle_context ioqueue_udpsock_t;
typedef struct ioqueue_handle_context ioqueue_tcpsock_t;
typedef struct ioqueue_handle_context ioqueue_acceptor_t;
typedef struct ioqueue_handle_context ioqueue_file_t;
typedef struct ioqueue_handle_context ioqueue_pipe_acceptor_t;

typedef struct ioqueue_overlapped_t { /* do not access directly */
	void* __internal[32];
}ioqueue_overlapped_t;

/**
 * callback while new client connects in 
 * @param user_data pass from async_accept
 * @param in_sock client socket, if you dont need this socket, 
 *		no need to call closesocket on it, just return zero!!!
 * @param err -1 on failed, 0 on success
 * @return non-zero means accept, zero reject
 */
typedef int (*ioqueue_on_accept_callback)(ioqueue_acceptor_t *acceptor, 
										   ioqueue_overlapped_t *overlapped,
										   SOCKET in_sock,
										   void *user_data, 
										   int err);

/**
 * create a new acceptor(TCP only)
 * @param [in] ioq io queue pointer 
 * @param [in] ip service ip address
 * @param [in] port service port
 * @param [out] acceptor pointer to result structure
 * @return 0 success, -1 failed
 */
TOOLKIT_API int ioqueue_acceptor_create(ioqueue_t *ioq, 
							const char *ip, 
							unsigned short port, 
							ioqueue_acceptor_t* acceptor);

/**
 * destory an acceptor
 */
TOOLKIT_API void ioqueue_acceptor_destroy(ioqueue_acceptor_t *acceptor);

/** close socket handler, so all pending async operation can be aborted */
TOOLKIT_API void ioqueue_acceptor_close(ioqueue_acceptor_t* acceptor);

/**
 * listen 
 * @return 0 ok, -1 failed
 */
TOOLKIT_API int ioqueue_acceptor_listen(ioqueue_acceptor_t* acceptor, int backlog);

/**
 * get raw socket handler
 */
TOOLKIT_API SOCKET ioqueue_acceptor_get_raw_socket(ioqueue_acceptor_t *sock);

/**
 * get owned ioqueue
 */
TOOLKIT_API ioqueue_t* ioqueue_acceptor_get_owned_ioqueue(ioqueue_acceptor_t* acceptor);

/**
 * set user customize data
 * @return old value
 */
TOOLKIT_API void *ioqueue_acceptor_set_user_data(ioqueue_acceptor_t* acceptor, void *user_data);

/**
 * get user customize data
 */
TOOLKIT_API void *ioqueue_acceptor_get_user_data(ioqueue_acceptor_t* acceptor);

/**
 * start a accept asynchronous operation
 * @param acceptor [in] pointer to acceptor
 * @param overlapped [in] operation based structure, cannot be free before callback invoked
 * @param on_accept_cb [in] accept callback, invoked after new socket accept
 * @return -1 failed, 0 on success
 */
TOOLKIT_API int ioqueue_acceptor_async_accept(ioqueue_acceptor_t* acceptor, 
								  ioqueue_overlapped_t *overlapped,
								  ioqueue_on_accept_callback on_accept_cb,
								  void *user_data);

/** blocked accept */
TOOLKIT_API int ioqueue_acceptor_accept(ioqueue_acceptor_t* acceptor, SOCKET *s, struct sockaddr *addr, int *addrlen, int timeout);

/**
 * get client tcp socket from acceptor and overlapped result, used in ioqueue_on_accept_callback!!!
 * @param acceptor [in] acceptor handle, from ioqueue_on_accept_callback params
 * @param s [in] acceptor returned client socket, from ioqueue_on_accept_callback params
 * @param tcpsock [out] result
 * @return -1 failed, 0 ok
 */
TOOLKIT_API int ioqueue_acceptor_create_client(ioqueue_acceptor_t* acceptor, SOCKET s, ioqueue_tcpsock_t *tcpsock);

/**
 * cancel all async operation
 * @return -1 failed, 0 ok
 */
TOOLKIT_API int ioqueue_acceptor_cancel(ioqueue_acceptor_t* acceptor);

////////////////////////////////////////////////////////////////////////////////////////////////
// tcp socket

/** asynchronous callbacks */

typedef void (*ioqueue_on_connect_callback)(ioqueue_tcpsock_t *tcpsock, 
											ioqueue_overlapped_t *overlapped,
											void *user_data, 
											int err);
typedef void (*ioqueue_on_send_callback)(ioqueue_tcpsock_t *tcpsock, 
										 ioqueue_overlapped_t *overlapped,
										 void *buf, 
										 unsigned int transfer_bytes, 
										 void *user_data,
										 int err);
typedef void (*ioqueue_on_recv_callback)(ioqueue_tcpsock_t *tcpsock, 
										 ioqueue_overlapped_t *overlapped,
										 void *buf, 
										 unsigned int transfer_bytes, 
										 void *user_data, 
										 int err);
typedef void (*ioqueue_on_recvuntil_callback)(ioqueue_tcpsock_t *tcpsock, 
											  ioqueue_overlapped_t *overlapped,
											  void *buf, 
											  unsigned int transfer_bytes, 
											  unsigned int header_length, 
											  void *user_data, 
											  int err);

/**
 * create tcp socket (non-connect, already bind to random local address)
 */
TOOLKIT_API int ioqueue_tcpsock_create(ioqueue_t *ioq, ioqueue_tcpsock_t *tcpsock);

/** create tcp socket object from socket handle */
TOOLKIT_API int ioqueue_tcpsock_create_from_handle(ioqueue_t *ioq, SOCKET s, ioqueue_tcpsock_t *tcpsock);

/**
 * connect to tcp(ip:port)
 * @param [in] ip remote server ip address
 * @param [in] port remote server port number, host byte order
 * @param [in] on_connect_cb callback function
 * @param [in] user_data pass to on_connect_cb
 * @return 0 success, -1 failed
 */
TOOLKIT_API int ioqueue_tcpsock_async_connect(ioqueue_tcpsock_t *tcpsock, 
								  ioqueue_overlapped_t *overlapped,
								  const char *ip, 
								  unsigned short port, 
								  ioqueue_on_connect_callback on_connect_cb,
								  void* user_data);

/** block connect */
TOOLKIT_API int ioqueue_tcpsock_conect(ioqueue_tcpsock_t *tcpsock, const char *ip, unsigned short port, int timeout);

/**
 * issue an asynchronous send operation
 * @param buf send buffer
 * @param len buffer length
 * @param on_send_cb callback funtion
 * @return -1 on failed, 0 on success
 */
TOOLKIT_API int ioqueue_tcpsock_async_sendsome(ioqueue_tcpsock_t *tcpsock, 
								   ioqueue_overlapped_t *overlapped,
								   void *buf, 
								   unsigned int len, 
								   ioqueue_on_send_callback on_send_cb,
								   void *user_data);
TOOLKIT_API int ioqueue_tcpsock_async_sendn(ioqueue_tcpsock_t *tcpsock, 
								ioqueue_overlapped_t *overlapped,
								void *buf, 
								unsigned int len,
								ioqueue_on_send_callback on_send_cb,
								void* user_data);
TOOLKIT_API int ioqueue_tcpsock_async_senduntil(ioqueue_tcpsock_t *tcpsock, 
									ioqueue_overlapped_t *overlapped,
									void *buf, 
									unsigned int len,
									const char *delimer,
									ioqueue_on_send_callback on_send_cb,
									void* user_data);

/** blocking send */
TOOLKIT_API int ioqueue_tcpsock_sendsome(ioqueue_tcpsock_t *tcpsock, 
							 void *buf, 
							 unsigned int len, 
							 int timeout);
TOOLKIT_API int ioqueue_tcpsock_sendn(ioqueue_tcpsock_t *tcpsock, 
						  void *buf, 
						  unsigned int len, 
						  int timeout);
TOOLKIT_API int ioqueue_tcpsock_senduntil(ioqueue_tcpsock_t *tcpsock, 
							  void *buf, 
							  unsigned int len, 
							  const char *delimer, 
							  int timeout);

/**
 * issue an asynchronous recv operation
 */
TOOLKIT_API int ioqueue_tcpsock_async_recvsome(ioqueue_tcpsock_t *tcpsock, 
								   ioqueue_overlapped_t *overlapped,
								   void *buf, 
								   unsigned int len, 
								   ioqueue_on_recv_callback on_recv_cb,
								   void *user_data);
TOOLKIT_API int ioqueue_tcpsock_async_recvn(ioqueue_tcpsock_t *tcpsock, 
								ioqueue_overlapped_t *overlapped,
								void *buf, 
								unsigned int len, 
								ioqueue_on_recv_callback on_recv_cb,
								void *user_data);
TOOLKIT_API int ioqueue_tcpsock_async_recvuntil(ioqueue_tcpsock_t *tcpsock, 
									ioqueue_overlapped_t *overlapped,
									void *buf, 
									unsigned int len,
									const char *delimer,
									ioqueue_on_recvuntil_callback on_recvuntil_cb,
									void *user_data);

/** blocking recv */
TOOLKIT_API int ioqueue_tcpsock_recvsome(ioqueue_tcpsock_t *tcpsock, 
							 void *buf, 
							 unsigned int len, 
							 int timeout);
TOOLKIT_API int ioqueue_tcpsock_recvn(ioqueue_tcpsock_t *tcpsock, 
						  void *buf, 
						  unsigned int len, 
						  int timeout);
TOOLKIT_API int ioqueue_tcpsock_recvuntil(ioqueue_tcpsock_t *tcpsock, 
							  void *buf, 
							  unsigned int len, 
							  const char *delimer, 
							  unsigned int *header_len,
							  int timeout);

/** close socket handler, so all pending async operation can be aborted */
TOOLKIT_API void ioqueue_tcpsock_close(ioqueue_tcpsock_t *tcpsock);

/**
 * destory an tcpsock
 */
TOOLKIT_API void ioqueue_tcpsock_destroy(ioqueue_tcpsock_t *tcpsock);

/**
 * shutdown wrapper
 */
TOOLKIT_API int ioqueue_tcpsock_shutdown(ioqueue_tcpsock_t *tcpsock, int how);

/**
 * get raw socket handler
 */
TOOLKIT_API SOCKET ioqueue_tcpsock_get_raw_socket(ioqueue_tcpsock_t *tcpsock);

/**
 * get owned ioqueue
 */
TOOLKIT_API ioqueue_t* ioqueue_tcpsock_get_owned_ioqueue(ioqueue_tcpsock_t *tcpsock);

/**
 * set user customize data
 * @return old value
 */
TOOLKIT_API void *ioqueue_tcpsock_set_user_data(ioqueue_tcpsock_t *tcpsock, void *user_data);

/**
 * get user customize data
 */
TOOLKIT_API void *ioqueue_tcpsock_get_user_data(ioqueue_tcpsock_t *tcpsock);

/**
 * cancel all async operation
 * @return -1 failed, 0 ok
 */
TOOLKIT_API int ioqueue_tcpsock_cancel(ioqueue_tcpsock_t* tcpsock);

////////////////////////////////////////////////////////////////////////////////////////////////
// udp socket

typedef void (*ioqueue_on_sendto_callback)(ioqueue_udpsock_t *udpsock, 
										   ioqueue_overlapped_t *overlapped,
										   void *buf, 
										   unsigned int transfer_bytes, 
										   void *user_data,
										   int err);
typedef void (*ioqueue_on_recvfrom_callback)(ioqueue_udpsock_t *udpsock, 
											 ioqueue_overlapped_t *overlapped,
											 const struct sockaddr *from,
											 int fromlen,
											 void *buf, 
											 unsigned int transfer_bytes, 
											 void *user_data,
											 int err);

/**
 * create udp socket
 */
TOOLKIT_API int ioqueue_udpsock_create(ioqueue_t *ioq, ioqueue_udpsock_t *udpsock);

/** create udp socket object from socket handle */
TOOLKIT_API int ioqueue_udpsock_create_from_handle(ioqueue_t *ioq, SOCKET s, ioqueue_udpsock_t *udpsock);

/** close socket handler, so all pending async operation can be aborted */
TOOLKIT_API void ioqueue_udpsock_close(ioqueue_udpsock_t *udpsock);

/**
 * destroy udp socket
 */
TOOLKIT_API void ioqueue_udpsock_destroy(ioqueue_udpsock_t *udpsock);

/**
 * issue an asynchronous sendto operation, mainly for udp udpsock
 */
TOOLKIT_API int ioqueue_udpsock_async_sendto(ioqueue_udpsock_t* sock, 
								 ioqueue_overlapped_t *overlapped,
								 void *buf, 
								 int len, 
								 const struct sockaddr* to, 
								 int tolen,
								 ioqueue_on_sendto_callback on_sendto_cb,
								 void *user_data);

/** blocking sendto */
TOOLKIT_API int ioqueue_udpsock_sendto(ioqueue_udpsock_t *udpsock, 
						   void *buf, 
						   int len, 
						   const struct sockaddr* to, 
						   int tolen, 
						   int timeout);

/**
 * issue an asynchronous recv operation, mainly for udp udpsock
 */
TOOLKIT_API int ioqueue_udpsock_async_recvfrom(ioqueue_udpsock_t* udpsock, 
								   ioqueue_overlapped_t *overlapped,
								   void* buf, 
								   int len, 
								   ioqueue_on_recvfrom_callback on_recvfrom_cb,
								   void *user_data);

/** blocking recvfrom */
TOOLKIT_API int ioqueue_udpsock_recvfrom(ioqueue_udpsock_t* udpsock, 
							 ioqueue_overlapped_t *overlapped,
							 void* buf, 
							 int len,
							 struct sockaddr *addr,
							 int *addrlen,
							 int timeout);

/**
 * get raw socket handler
 */
TOOLKIT_API SOCKET ioqueue_udpsock_get_raw_socket(ioqueue_udpsock_t *udpsock);

/**
 * get owned ioqueue
 */
TOOLKIT_API ioqueue_t* ioqueue_udpsock_get_owned_ioqueue(ioqueue_udpsock_t *udpsock);

/**
 * set user customize data
 * @return old value
 */
TOOLKIT_API void *ioqueue_udpsock_set_user_data(ioqueue_udpsock_t *udpsock, void *user_data);

/**
 * get user customize data
 */
TOOLKIT_API void *ioqueue_udpsock_get_user_data(ioqueue_udpsock_t *udpsock);

/**
 * cancel all async operation
 * @return -1 failed, 0 ok
 */
TOOLKIT_API int ioqueue_udpsock_cancel(ioqueue_udpsock_t *udpsock);

////////////////////////////////////////////////////////////////////////////////////////////////
// file

/** asychronous callbacks */
typedef void (*ioqueue_on_read_callback)(ioqueue_file_t* file, 
										 ioqueue_overlapped_t *overlapped,
										 void *buf, 
										 unsigned int transfer_bytes, 
										 void *user_data,
										 int err);
typedef void (*ioqueue_on_readuntil_callback)(ioqueue_file_t* file, 
											  ioqueue_overlapped_t *overlapped,
											  void *buf, 
											  unsigned int transfer_bytes, 
											  unsigned int header_length,
											  void *user_data,
											  int err);
typedef void (*ioqueue_on_write_callback)(ioqueue_file_t* file, 
										  ioqueue_overlapped_t *overlapped,
										  void *buf,
										  unsigned int transfer_bytes, 
										  void *user_data,
										  int err);

/**
 * create file from path
 * @param path local file system path
 * @param dwDesiredAccess refer CreateFile api
 * @param dwShareMode refer CreateFile api
 * @param dwCreationDisposition refer CreateFile api
 * @param dwFlagsAndAttributes refer CreateFile api
 * @return 0 success, -1 failed
 */
TOOLKIT_API int ioqueue_file_create(ioqueue_t *ioq,
						const char *path, 
						DWORD dwDesiredAccess,
						DWORD dwShareMode,
						DWORD dwCreationDisposition, 
						DWORD dwFlagsAndAttributes,
						ioqueue_file_t *file);

TOOLKIT_API int ioqueue_file_create_from_handle(ioqueue_t *ioq, HANDLE h, ioqueue_file_t *file);

/** close file handler, so all pending async operation can be aborted */
TOOLKIT_API void ioqueue_file_close(ioqueue_file_t* file);

/**
 * destroy file 
 */
TOOLKIT_API void ioqueue_file_destroy(ioqueue_file_t* file);

/**
 * issue an asynchronous read operation
 * @return -1 on failed, 0 on success
 */
TOOLKIT_API int ioqueue_file_async_readsome(ioqueue_file_t* file, 
								ioqueue_overlapped_t *overlapped,
								void *buf, 
								unsigned int len, 
								ioqueue_on_read_callback on_read_cb,
								void *user_data);
TOOLKIT_API int ioqueue_file_async_readn(ioqueue_file_t* file, 
							 ioqueue_overlapped_t *overlapped,
							 void *buf, 
							 unsigned int len, 
							 ioqueue_on_read_callback on_read_cb,
							 void *user_data);

/** blocking read */
TOOLKIT_API int ioqueue_file_readsome(ioqueue_file_t *file, void *buf, unsigned int len);
TOOLKIT_API int ioqueue_file_readn(ioqueue_file_t *file, void *buf, unsigned int len);

/**
 * issue an asynchronous read at operation
 * @param buf read buffer
 * @param len buffer length
 * @param pos file read position 
 * @param on_read_cb callback function
 */
TOOLKIT_API int ioqueue_file_async_readsome_at(ioqueue_file_t* file, 
							   ioqueue_overlapped_t *overlapped,
							   void *buf, 
							   unsigned int len, 
							   DWORD posLow, 
							   DWORD posHigh, 
							   ioqueue_on_read_callback on_read_cb,
							   void *user_data);
TOOLKIT_API int ioqueue_file_async_readn_at(ioqueue_file_t* file, 
								ioqueue_overlapped_t *overlapped,
								void *buf, 
								unsigned int len, 
								DWORD posLow, 
								DWORD posHigh, 
								ioqueue_on_read_callback on_read_cb,
								void *user_data);


/** blocking read at */
TOOLKIT_API int ioqueue_file_readsome_at(ioqueue_file_t *file, void *buf, unsigned int len, DWORD posLow, DWORD posHigh);
TOOLKIT_API int ioqueue_file_readn_at(ioqueue_file_t *file, void *buf, unsigned int len, DWORD posLow, DWORD posHigh);

/**
 * issue an asynchronous write at operation
 */
TOOLKIT_API int ioqueue_file_async_writesome(ioqueue_file_t* file, 
								 ioqueue_overlapped_t *overlapped,
								 void* buf, 
								 unsigned int len, 
								 ioqueue_on_write_callback on_write_cb,
								 void *user_data);
TOOLKIT_API int ioqueue_file_async_writen(ioqueue_file_t* file, 
							  ioqueue_overlapped_t *overlapped,
							  void* buf, 
							  unsigned int len, 
							  ioqueue_on_write_callback on_write_cb,
							  void *user_data);

/** blocking write */
TOOLKIT_API int ioqueue_file_writesome(ioqueue_file_t* file, const void *buf, unsigned int len);
TOOLKIT_API int ioqueue_file_writen(ioqueue_file_t* file, const void *buf, unsigned int len);

/**
 * issue an asynchronous write at operation
 */
TOOLKIT_API int ioqueue_file_async_writesome_at(ioqueue_file_t* file, 
									ioqueue_overlapped_t *overlapped,
									void* buf, 
									unsigned int len, 
									DWORD posLow, 
									DWORD posHigh, 
									ioqueue_on_write_callback on_write_cb,
									void *user_data);
TOOLKIT_API int ioqueue_file_async_writen_at(ioqueue_file_t* file, 
								 ioqueue_overlapped_t *overlapped,
								 void* buf, 
								 unsigned int len, 
								 DWORD posLow, 
								 DWORD posHigh, 
								 ioqueue_on_write_callback on_write_cb,
								 void *user_data);

/** blocking write at */
TOOLKIT_API int ioqueue_file_writesome_at(ioqueue_file_t* file, 
							  const void *buf, 
							  unsigned int len, 
							  DWORD posLow, 
							  DWORD posHigh);
TOOLKIT_API int ioqueue_file_writen_at(ioqueue_file_t* file, 
						   const void *buf, 
						   unsigned int len, 
						   DWORD posLow, 
						   DWORD posHigh);

/**
 * get owned ioqueue
 */
TOOLKIT_API ioqueue_t* ioqueue_file_get_owned_ioqueue(ioqueue_file_t* file);

/**
 * get inner file handler
 */
TOOLKIT_API HANDLE ioqueue_file_get_raw_handle(ioqueue_file_t* file);

/**
 * set user customized data
 * @return old value
 */
TOOLKIT_API void *ioqueue_file_set_user_data(ioqueue_file_t* file, void* user_data);

/** 
 * get user customized data
 */
TOOLKIT_API void *ioqueue_file_get_user_data(ioqueue_file_t* file);

/**
 * cancel all async operation
 * @return -1 failed, 0 ok
 */
TOOLKIT_API int ioqueue_file_cancel(ioqueue_file_t* file);

////////////////////////////////////////////////////////////////////////////////////////////////
// pipe server

/**
 * callback while new pipe client connects in 
 * @param user_data pass from async_accept
 * @param in_sock client socket, if you dont need this socket, 
 *		no need to call closesocket on it, just return zero!!!
 * @param err -1 on failed, 0 on success
 * @return non-zero means accept, zero reject
 */
typedef int (*ioqueue_on_pipe_accept_callback)(ioqueue_pipe_acceptor_t *acceptor, 
											   ioqueue_overlapped_t *overlapped,
											   HANDLE pipe,
											   void *user_data, 
											   int err);

/**
 * pipe acceptor
 */
TOOLKIT_API int ioqueue_pipe_acceptor_create(ioqueue_t *ioq, const char *name, ioqueue_pipe_acceptor_t *acceptor);

/**
 * destroy pipe acceptor
 */
TOOLKIT_API void ioqueue_pipe_acceptor_destroy(ioqueue_pipe_acceptor_t *acceptor);

/**
 * get owned ioqueue
 */
TOOLKIT_API ioqueue_t* ioqueue_pipe_acceptor_get_owned_ioqueue(ioqueue_pipe_acceptor_t *acceptor);

/**
 * set user customize data
 * @return old value
 */
TOOLKIT_API void *ioqueue_pipe_acceptor_set_user_data(ioqueue_pipe_acceptor_t *acceptor, void *user_data);

/**
 * get user customize data
 */
TOOLKIT_API void *ioqueue_pipe_acceptor_get_user_data(ioqueue_pipe_acceptor_t *acceptor);

/** async accept */
TOOLKIT_API int ioqueue_pipe_acceptor_async_accept(ioqueue_pipe_acceptor_t *acceptor,
									   ioqueue_overlapped_t *overlapped,
									   ioqueue_on_pipe_accept_callback cb,
									   void *user_data);

/** blocked accept */
TOOLKIT_API int ioqueue_pipe_acceptor_accept(ioqueue_pipe_acceptor_t *acceptor, HANDLE *p_pipe, int timeout);

/**
 * get client pipe from acceptor and overlapped result, used in ioqueue_on_pipe_accept_callback!!!
 * @param acceptor [in] acceptor handle, from ioqueue_on_accept_callback params
 * @param h [in] acceptor returned client pipe handle, from ioqueue_on_pipe_accept_callback params
 * @param pipe [out] result
 * @return -1 failed, 0 ok
 */
TOOLKIT_API int ioqueue_pipe_acceptor_create_client(ioqueue_pipe_acceptor_t *acceptor, HANDLE h, ioqueue_file_t *pipe);

/**
 * cancel all async operation
 * @return -1 failed, 0 ok
 */
TOOLKIT_API int ioqueue_pipe_acceptor_cancel(ioqueue_pipe_acceptor_t *acceptor);
TOOLKIT_API int ioqueue_pipe_acceptor_close_pending_handle(ioqueue_pipe_acceptor_t *acceptor);


#ifndef _WIN32

TOOLKIT_API int ioqueue_overlapped_get_mask(const ioqueue_overlapped_t* const io);
TOOLKIT_API void ioqueue_overlapped_set_mask(ioqueue_overlapped_t* io, int mask_value);
TOOLKIT_API int ioqueue_overlapped_get_type(const ioqueue_overlapped_t* const io);

#endif //NOT _WIN32


#ifdef __cplusplus
} // extern "C" {
#endif

#endif // IOQUEUE_H