rpc_over_ipc_nano.cpp

/*
	Copyright (c) 2012 Martin Sustrik  All rights reserved.

	Permission is hereby granted, free of charge, to any person obtaining a copy
	of this software and associated documentation files (the "Software"),
	to deal in the Software without restriction, including without limitation
	the rights to use, copy, modify, merge, publish, distribute, sublicense,
	and/or sell copies of the Software, and to permit persons to whom
	the Software is furnished to do so, subject to the following conditions:

	The above copyright notice and this permission notice shall be included
	in all copies or substantial portions of the Software.

	THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
	IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
	FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
	THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
	LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
	FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
	IN THE SOFTWARE.
*/

#include "../src/nn.h"
#include "../src/pair.h"
#include "../src/pubsub.h"
#include "../src/ipc.h"

#include "testutil.h"
#include <stdio.h>
/*  Tests IPC transport. */


class MsgBuffer {
public:
	char *data = NULL;
	int size = 0;
	int pos = 0;
	int is_nanobuffer = 0;
	MsgBuffer() {
		// probably calling ReadNanoSocket later, which sets up data and length then
		// but Free() is special then, since it needs to call nn_free()
	}
	MsgBuffer(int size_) {
		data = (char *)malloc(size_);
		size = size_;
	}
	~MsgBuffer() {
		Free();
	}
	void WriteInt(int tmp) {
		data[pos] = tmp;
		pos += 4;
	}
	int ReadInt() {
		int ret = data[pos];
		pos += 4;
		return ret;
	}
	int Seek(int pos_) {
		pos = pos_;
	}
	int Free() {
		//printf("Free()\n");
		if(data == NULL) {
			//printf("MsgBuffer is already free'd\n");
			return 0;
		}
		if(!is_nanobuffer) {

			free(data);
		} else {
			nn_freemsg(data);
		}
		data = NULL;
		return 1;
	}
	int SendNanoSocket(int socket) {
		int ret = nn_send(socket, data, size, 0);

		return ret;
	}
	int ReadNanoSocket(int socket, int block = 1) {
		is_nanobuffer = 1;
		// we dont know the size of the answer, so let nanomsg give us whatever the server send: http://nanomsg.org/v0.1/nn_recv.3.html
		int ret = nn_recv(socket, &data, /*len*/NN_MSG, block ? 0 : NN_DONTWAIT);
		pos = 0;
		size = ret; // well, not on errors, todo
		//printf("ret ReadNanoSocket: %d\n", ret);
		return ret;
	}
};

class Peer {
public:
	int socket = 0;
	Peer() {
		int family = AF_SP;
		int protocol = NN_PAIR;
		socket = nn_socket(family, protocol);
		if(socket == -1) {
			fprintf(stderr, "Failed create socket: %s [%d]\n",
				nn_err_strerror(errno),
				(int)errno);
			nn_err_abort();
		}

	}

	int Connect(char *address) {
		int rc = nn_connect(socket, address);
		if(rc < 0) {
			fprintf(stderr, "Failed connect to \"%s\": %s [%d] \n",
				address,
				nn_err_strerror(errno),
				(int)errno
			);
			nn_err_abort();
		}
		return rc;
	}


	int Bind(char *address) {
		int rc;
		rc = nn_bind(socket, address);
		if(rc < 0) {
			fprintf(stderr, "Failed bind to \"%s\": %s [%d]\n",
				address,
				nn_err_strerror(errno),
				(int)errno
			);
			nn_err_abort();
		}
		return rc;
	}

	int SendBuffer(MsgBuffer *buffer) {
		return buffer->SendNanoSocket(socket);
	}
	int ReceiveBuffer(MsgBuffer *buffer, int block = 1) {
		return buffer->ReadNanoSocket(socket, block);
	}
};


typedef enum { RPC_ADD, RPC_PRINT } RPC_REQUEST;

#define SOCKET_ADDRESS "ipc://test.ipc"
class RPC : public Peer {
public:
	char *name;
	RPC(char *name_) {
		name = name_;
	}
	int Dispatch() {
		MsgBuffer query;
		int recv_ret = ReceiveBuffer(&query, 0);
		// nothing to process here
		if (recv_ret < 0)
			return 0;
		int rpc_id = query.ReadInt();

		//printf("recv_ret=%d rpc: id=%d\n", recv_ret, rpc_id);
		switch(rpc_id) {
		case RPC_ADD: {
			int a = query.ReadInt();
			int b = query.ReadInt();
			int ret = a + b;

			MsgBuffer answer(4);
			answer.WriteInt(a + b);
			SendBuffer(&answer);


			break;
		}
		}
		return 765;
	}
	int Add(int a, int b) {

		MsgBuffer query(12);
		query.WriteInt(RPC_ADD);
		query.WriteInt(50);
		query.WriteInt(3);
		//printf(" %d %d %d <<\n", query.data[0], query.data[4], query.data[8]);

		int ret = SendBuffer(&query);
		//printf("nn_send ret=%d\n", ret);

		MsgBuffer answer2;
		ReceiveBuffer(&answer2);
		return answer2.ReadInt();
	}
};


// cd C:\Users\kung\Desktop\nanomsg-1.0.0\nanomsg-1.0.0\build\Debug
// server: nanocat --bind-ipc foo --pair --format ascii
// client: nanocat --connect-ipc foo --pair  --data wuath

DWORD WINAPI func_server(void *Param) {

	RPC server("Server");
	server.Bind(SOCKET_ADDRESS);

	int errors = 0;
	for (int i=0; i<10; i++) {
		server.Dispatch();
		int ret = server.Add(50, 3);
		printf("Server> a + b = %d\n", ret);
		if (ret != 53) {
			printf("ERROR %d\n", i);
			errors++;
		}
	}
	printf("Server: done errors=%d", errors);


	return 123;
}

DWORD WINAPI func_client(void *Param) {

	RPC client("Client");
	client.Connect(SOCKET_ADDRESS);
	
	int errors = 0;
	for (int i=0; i<10; i++) {
		client.Dispatch();
		int ret = client.Add(50, 3);
		printf("Client> a + b = %d\n", ret);
		if (ret != 53) {
			printf("ERROR %d\n", i);
		}
	}
	
	printf("Client: done errors=%d", errors);
	//while(1) {
	//	client.Dispatch();
	//}



	return 456;
}

int meh()
{
	DWORD thread_server, thread_client;
	CreateThread(NULL, 0, func_server, (void*)NULL, 0, &thread_server);
	CreateThread(NULL, 0, func_client, (void*)NULL, 0, &thread_client);

	/*
	rc = nn_recv (sock, buf, data_len+1, 0);
	if (rc < 0) {
		fprintf (stderr, "Failed to recv: %s [%d] (%s:%d)\n",
			nn_err_strerror (errno),
			(int) errno, file, line);
		nn_err_abort ();
	}
	if (rc != (int)data_len) {
		fprintf (stderr, "Received data has wrong length: %d != %d (%s:%d)\n",
			rc, (int) data_len,
			file, line);
		nn_err_abort ();
	}
	######
	rc = nn_send (sock, data, data_len, 0);
	if (rc < 0) {
		fprintf (stderr, "Failed to send: %s [%d] (%s:%d)\n",
			nn_err_strerror (errno),
			(int) errno, file, line);
		nn_err_abort ();
	}
	if (rc != (int)data_len) {
		fprintf (stderr, "Data to send is truncated: %d != %d (%s:%d)\n",
			rc, (int) data_len,
			file, line);
		nn_err_abort ();
	}
	*/

	//test_close(sb);

	return 0;
}

// test c++ object deconstructor
int main() {
	int ret = meh();
	getchar();
	return ret;
}