async redis with G-WAN (work in progress now)

async-test.c

// ============================================================================
// C servlet sample for the G-WAN Web Application Server (http://trustleap.ch/)
// ----------------------------------------------------------------------------
// async-test.c: G-WAN async operations test
// ============================================================================
#include "gwan.h" // G-WAN exported functions
#include <hiredis/adapters/libev.h>
#include <hiredis/async.h>
#include <unistd.h>
#include <string.h>
#include "anet.c"

#pragma link "hiredis"
#pragma link "ev"

typedef struct 
{ 
	kv_t *kv;   // a Key-Value store
	char *blah; // a string
	int   val;  // a counter
	struct ev_loop *loop;
	ev_async async;
	ev_periodic periodic;
	void (*r_cb)(EV_P_ struct ev_io *r, int revents);
	void (*w_cb)(EV_P_ struct ev_io *w, int revents);
} data_t;

void getCallback(redisAsyncContext *c, void *r, void *privdata) {
	redisReply *reply = r;
	if (reply == NULL) return;
	printf("argv[%s]: %s\n", (char*)privdata, reply->str);

	/* Disconnect after receiving the reply to GET */
	redisAsyncDisconnect(c);
}

void connectCallback(const redisAsyncContext *c, int status) {
	if (status != REDIS_OK) {
		printf("Error: %s\n", c->errstr);
		return;
	}
	printf("Connected...\n");
}

void disconnectCallback(const redisAsyncContext *c, int status) {
	if (status != REDIS_OK) {
		printf("Error: %s\n", c->errstr);
		return;
	}
	printf("Disconnected...\n");
}

int main(int argc, char *argv[])
{
	data_t **Data = (data_t**)get_env(argv, US_SERVER_DATA),
			*data;
	data = *Data;

	xbuf_t *reply = get_reply(argv);

	void **r_d = (void**)get_env(argv, US_REQUEST_DATA);

	ev_async_send(data->loop, &data->async);

	if(!r_d[0]) // we did not setup our per-request counter yet
	{
		r_d[0] = (void*)1; // use the persistent pointer as a simple counter

		char head[] = "HTTP/1.1 200 OK\r\n"
					  "Connection: close\r\n"
					  "Content-length: 60\r\n" // 3 * 20 characters
					  "Content-type: text/plain; charset=utf-8\r\n"
					  "\r\n\r\n";
		xbuf_ncat(reply, head, sizeof(head) - 1);


		char err[256];
		int fd = anetTcpNonBlockConnect(err, "127.0.0.1", 8080);
		ev_io r;
		ev_io w;
		ev_io_init(&r, data->r_cb, fd, EV_READ);
		ev_io_init(&w, data->w_cb, fd, EV_WRITE);

		ev_io_start(data->loop, &w);
		ev_io_start(data->loop, &r);


		// ----------------------------------------------------------------------
		// tell G-WAN when to run this script again (for the same request)
		// ----------------------------------------------------------------------
		//wake_up(argv, 1000, WK_MS); // WK_MS:milliseconds, WK_FD:file_descriptor
		wake_up(argv, fd, WK_FD, NULL);


		redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379);
		if (c->err) {
			/* Let *c leak for now... */
			printf("Error: %s\n", c->errstr);
			return 1;
		}

		redisLibevAttach(data->loop, c);
		redisAsyncSetConnectCallback(c, connectCallback);
		redisAsyncSetDisconnectCallback(c, disconnectCallback);
		redisAsyncCommand(c, NULL, NULL, "SET key %b", argv[argc-1], strlen(argv[argc-1]));
		redisAsyncCommand(c, getCallback, (char*)"end-1", "GET key");
	} else {
		static char msg[] = "Hello, ANSI C!";
		xbuf_ncat(get_reply(argv), msg, sizeof(msg) - 1);
	}

	//sleep(1);

	return RC_NOHEADERS + RC_STREAMING; 

	//return 200; // return an HTTP code (200:'OK')
}
// ============================================================================
// End of Source Code
// ============================================================================

main.c

#pragma debug     // uncomment to get symbols/line numbers in crash reports

#include "gwan.h"   // G-WAN exported functions

#include <stdio.h>  // printf()
#include <string.h> // strcmp()
#include <malloc.h> // calloc()
#include <pthread.h>
#include <ev.h>
#pragma link "ev"
#pragma link "pthread"


void *worker(void *arg);
void periodic_cb(EV_P_ struct ev_periodic *w, int revents);
ev_tstamp reschedule_cb(struct ev_periodic *w, ev_tstamp now);
void async_cb(EV_P_ struct ev_async *w, int revents)

// ----------------------------------------------------------------------------
// structure holding pointers for persistence
// ----------------------------------------------------------------------------
typedef struct 
{ 
	kv_t *kv;   // a Key-Value store
	char *blah; // a string
	int   val;  // a counter
	struct ev_loop *loop;
	ev_async async;
	ev_periodic periodic;
} data_t;

// ----------------------------------------------------------------------------
// init() will initialize your data structures, load your files, etc.
// ----------------------------------------------------------------------------
// init() should return -1 if failure (to allocate memory for example)
int init(int argc, char *argv[])
{
	// get the Handler persistent pointer to attach anything you need
	//data_t **data = (data_t**)get_env(argv, US_HANDLER_DATA);
	//data_t **data = (data_t**)get_env(argv, US_VHOST_DATA);
	data_t **Data = (data_t**)get_env(argv, US_SERVER_DATA),
			*data;

	// initialize the persistent pointer by allocating our structure
	// attach structures, lists, sockets with a back-end/database server, 
	// file descriptiors for custom log files, etc.
	*Data = (data_t*)calloc(1, sizeof(data_t));
	if(!*Data)
		return -1; // out of memory
	data = *Data;

	// define which handler states we want to be notified in main():
	// enum HANDLER_ACT { 
	//  HDL_INIT = 0, 
	//  HDL_AFTER_ACCEPT, // just after accept (only client IP address setup)
	//  HDL_AFTER_READ,   // each time a read was done until HTTP request OK
	//  HDL_BEFORE_PARSE, // HTTP verb/URI validated but HTTP headers are not 
	//  HDL_AFTER_PARSE,  // HTTP headers validated, ready to build reply
	//  HDL_BEFORE_WRITE, // after a reply was built, but before it is sent
	//  HDL_HTTP_ERRORS,  // when G-WAN is going to reply with an HTTP error
	//  HDL_CLEANUP };
	/*	u32 *states = (u32 *)get_env(argv, US_HANDLER_STATES);
	*states = (1L << HDL_AFTER_ACCEPT) 
			| (1L << HDL_AFTER_READ)
			| (1L << HDL_AFTER_PARSE)
			| (1L << HDL_BEFORE_WRITE);
	*/

	// initialize libev loop
	data->loop = ev_loop_new(0);
	ev_async_init(&data->async, async_cb);
	ev_periodic_init(&data->periodic, periodic_cb, 0, 1000, reschedule_cb);

	// start worker thread
	pthread_t thread;

	pthread_create (&thread, NULL, worker, data);
	return 0;
}

// ----------------------------------------------------------------------------
// clean() will free any allocated memory and possibly log summarized stats
// ----------------------------------------------------------------------------
void clean(int argc, char *argv[])
{
	// free any data attached to your persistence pointer
	data_t **data = (data_t**)get_env(argv, US_SERVER_DATA);

	// we could close a data->log custom file 
	// if the structure had a FILE *log; field
	// fclose(data->log);

  // FIXME
	if(data)
		free(*data);
}

void async_cb(EV_P_ struct ev_async *w, int revents)
{
	printf("fn: %s\n", __func__);
}

void periodic_cb(EV_P_ struct ev_periodic *w, int revents)
{
	printf("fn: %s\n", __func__);
}

ev_tstamp reschedule_cb(struct ev_periodic *w, ev_tstamp now)
{
	printf("fn: %s\n", __func__);
	return ev_time() + 1000;
}


void *worker(void *arg)
{
	printf("yan işlem çalışıyor...\n");
	data_t *data = (data_t *)arg;

	ev_async_start(data->loop, &data->async);
	ev_periodic_start(data->loop, &data->periodic);

	ev_run(data->loop, 0);
	return NULL;
}

int main(int argc, char *argv[])
{
  return 255;
}