Corrected networking examples

changed the api for interacting with sockets, sockets
now have a callback, `socket:receive(function(stream) ... end)`, which
they can use to decide what to do when called.

Sockets also have a block:recv() function, which will block EVERYTHING
until the socket receives data. This should probably not be used.

1 files changed, 272 insertions, 84 deletions

diff --git a/src/shared/lua_api/load_net.cpp b/src/shared/lua_api/load_net.cpp
index 4b8475f..b6e95bc 100644
--- a/src/shared/lua_api/load_net.cpp
+++ b/src/shared/lua_api/load_net.cpp
@@ -7,9 +7,10 @@ The number values below are only for refrence. You should use net.PAIR, net.BUS,
 ect.
 @module net
 @usage
-	local s = net.newsocket()
+	--Server
+	local s = net.newsocket(net.REP)
 	s:bind("tcp://127.0.0.1:8765")
-	s:receive("ping",function(stream)
+	s:receive(function(stream)
 		local message = stream:readstring()
 		s:send("pong",function(stream)
 			stream:writestring(message .. " world!")
@@ -17,11 +18,13 @@ ect.
 	end)
 
 @usage
-	local c = net.newsocket()
-	c:receive("pong",function(stream)
+	--Client
+	local c = net.newsocket(net.REQ)
+	c:connect("tcp://127.0.0.1:8765")
+	c:receive(function(stream)
 		print(stream:readstring())
 	end)
-	c:send("ping",function(stream)
+	c:send(function(stream)
 		stream:writestring("Hello,")
 	end)
 */
@@ -36,27 +39,29 @@ extern "C" {
 #include <string>
 #include <map>
 
-#include <nng.h>
-#include <transport/inproc/inproc.h>
-#include <transport/ipc/ipc.h>
-#include <transport/tcp/tcp.h>
-#include <transport/tls/tls.h>
-#include <transport/zerotier/zerotier.h>
-
-#include <protocol/pair1/pair.h>
-#include <protocol/bus0/bus.h>
-#include <protocol/pubsub0/pub.h>
-#include <protocol/pubsub0/sub.h>
-#include <protocol/pipeline0/pull.h>
-#include <protocol/pipeline0/push.h>
-#include <protocol/reqrep0/req.h>
-#include <protocol/reqrep0/rep.h>
-#include <protocol/survey0/respond.h>
-#include <protocol/survey0/survey.h>
+#include <nng/nng.h>
+
+#include <nng/transport/inproc/inproc.h>
+#include <nng/transport/ipc/ipc.h>
+#include <nng/transport/tcp/tcp.h>
+#include <nng/transport/tls/tls.h>
+#include <nng/transport/zerotier/zerotier.h>
+
+#include <nng/protocol/pair1/pair.h>
+#include <nng/protocol/bus0/bus.h>
+#include <nng/protocol/pubsub0/pub.h>
+#include <nng/protocol/pubsub0/sub.h>
+#include <nng/protocol/pipeline0/pull.h>
+#include <nng/protocol/pipeline0/push.h>
+#include <nng/protocol/reqrep0/req.h>
+#include <nng/protocol/reqrep0/rep.h>
+#include <nng/protocol/survey0/respond.h>
+#include <nng/protocol/survey0/survey.h>
 
 #include "load_net.hpp"
 
 #include <shared/util/hashmap.hpp>
+#include <shared/lua_api/common.hpp>
 #include "stream.hpp"
 
 /***
@@ -72,8 +77,18 @@ it just sent.
 */
 
 /***
+Publish protocol.
+The first half of the pub/sub protocol.
 @field PUB 3
+*/
+
+/***
+Subscribe protocol
+The second half of the pub/sub protocol.
 @field SUB 4
+*/
+
+/***
 @field PULL 5
 @field PUSH 6
 @field REQ 7
@@ -92,8 +107,6 @@ it just sent.
 #define RESPOND 9
 #define SURVEY 10
 
-std::map<nng_socket*,int> netfuncs;
-
 //Some defines for things that the library dosn't have
 
 
@@ -277,49 +290,186 @@ int lstream_writestring(lua_State* L){
 	return 0;
 }
 
-void gameloop_net(lua_State* L){
-	//printf("Doing net of gameloop\n");
-	for(std::map<nng_socket*,int>::iterator it = netfuncs.begin(); it != netfuncs.end(); ++it){
-		//printf("In for socket %d\n");
-		//char* buf = NULL;
-		//size_t size;
-		//int err = nng_recv(*(it->first), &buf, &size, NNG_FLAG_NONBLOCK | NNG_FLAG_ALLOC);
-		nng_msg *msgp;
-		int err = nng_recvmsg(*(it->first), &msgp, NNG_FLAG_NONBLOCK);
+bool socket_can_receive(int type){
+	switch(type){
+		case PAIR:
+		case BUS:
+		case SUB:
+		case PULL:
+		case REQ:
+		case REP:
+		case RESPOND:
+		case SURVEY:
+			return true;
+		default:
+			return false;
+	}
+}
+
+//socket:block_recv()
+int block_recv(lua_State *L){//{socket}
+	lua_getfield(L,-1,"fd");//{socket},ud_socket
+	nng_socket *socket = (nng_socket*)lua_touserdata(L,-1);
+	lua_pop(L,2);//
+	nng_msg *msgp;
+	int err = nng_recvmsg(*socket,&msgp,0);
+	if(err){
+		printf("Net error: %s\n\t\n",nng_strerror(err));
+		lua_pushstring(L,"Error while receving message:");//Err
+		lua_pushstring(L,nng_strerror(err));//Err,msg
+		lua_concat(L,2);//Full_err
+		lua_error(L);
+	}else{
 		char* buf = (char*)nng_msg_body(msgp);
 		size_t size = nng_msg_len(msgp);
-		//printf("Got bytes: %d EAGAIN:%d\n",bytes,EAGAIN);
-		if(err != 0 && err != NNG_EAGAIN){
-			printf("Net error: %s\n\t err: %d\n\teagain:%d\n",nng_strerror(err),err,NNG_EAGAIN);
-			//lua_pushstring(L,"Failed to receive");
-			//lua_error(L);
-		}else if(err == NNG_EAGAIN){
-			//do nothing
+		struct stream* stream = stream_create();
+		stream->length = size;
+		stream->data = buf;
+		//stream_print(stream);
+		lua_newtable(L);//{}
+		lua_pushlightuserdata(L,stream);//{},ud_stream
+		lua_setfield(L,-2,"data");//{data=stream}
+		luaL_getmetatable(L,"net.stream");//{data=stream},{net.stream}
+		lua_setmetatable(L,-2);//{stream}
+	}
+	return 1;
+}
+
+void gameloop_net(lua_State* L){
+	//printf("Doing net of gameloop,starting with %d args\n",lua_gettop(L));
+	//printf("Got net\n");
+	lua_getglobal(L,"net");//{net}
+	lua_getfield(L,-1,"sockets");//{net},{sockets}
+	lua_pushnil(L);//{net},{sockets},nil
+	//printf("Found sockets\n");
+	while(lua_next(L,-2) != 0){
+		//printf("Got first socket value\n");
+		//{net},{sockets},{socket},true
+		//printf("%s - %s\n",lua_typename(L,lua_type(L,-2)),lua_typename(L,lua_type(L,-1)));
+		lua_getfield(L,-2,"fd");//{net},{sockets},{socket},true,fd
+		nng_socket *socket = (nng_socket*)lua_touserdata(L,-1);
+		lua_getfield(L,-3,"n");//{net},{sockets},{socket},true,fd,type
+		int stype = lua_tonumber(L,-1);
+		//printf("Got a socket of type %d\n",stype);
+		lua_pop(L,3);//{net},{sockets},{socket}//Keep key for next iteration of lua_next()
+		if(!socket_can_receive(stype)){
+			//printf("Socket cannot receive, breaking!\n");
+			continue;
+		}
+		//printf("About to push errorfunc\n");
+		pusherrorfunc(L);//{net},{sockets},{socket},errfunc()
+		//printf("Done pushing errorfunc\n");
+		lua_getfield(L,-2,"receive");//{net},{sockets},{socket},errfunc(),(socket.receive | nil)
+		//printf("Got field\n");
+		if(lua_isnil(L,-1)){//Make sure we have a receive 
+			//printf("Listen-able socket type %d has no .receive method\n",stype);
+			lua_pop(L,2);
+			continue;
+		}
+		//{net},{sockets},{socket},socket.receive
+		//printf("Right before recv\n");
+		nng_msg *msgp;
+		//printf("About to recvmsg(), socket: %p, msgp: %p\n",socket,msgp);
+		int err = nng_recvmsg(*socket,&msgp,NNG_FLAG_NONBLOCK);
+		//size_t recvsize;
+		//char *buf;
+		//int err = nng_recv(*socket,&buf,&recvsize,NNG_FLAG_NONBLOCK);
+		//printf("Done with recvmsg() err: %d:\n",err);
+		if(err){
+			switch(err){
+				case NNG_EAGAIN: break;
+				case NNG_ESTATE:
+					if(stype == REQ) break;
+				default:
+					printf("Net error: %s\n\tSocket type:%d\n",nng_strerror(err),stype);
+					lua_pushstring(L,"Error while receving message:");
+					lua_pushstring(L,nng_strerror(err));
+					lua_concat(L,2);
+					lua_error(L);
+			}
+			lua_pop(L,2);
 		}else{
-			//find how long until the first null character
+			//printf("Actually receving message\n");
+			char* buf = (char*)nng_msg_body(msgp);
+			//printf("Got message body\n");
+			size_t size = nng_msg_len(msgp);
+			//size_t size = recvsize;
+			//printf("Got mesage body\n");
 			struct stream* stream = stream_create();
 			stream->length = size;
 			stream->data = buf;
-			stream_print(stream);
-			int f = it->second;
-			if(f == -1){//not set yet
-				lua_pushstring(L,"Got message for a socket without a receiving function");
-				lua_error(L);
-			}
-			lua_rawgeti(L,LUA_REGISTRYINDEX,f);//function(stream)
-			lua_newtable(L);//function(stream),{}
-			lua_pushlightuserdata(L,stream);//func,{},ud_stream
-			lua_setfield(L,-2,"data");//func,{data=stream}
-			luaL_getmetatable(L,"net.stream");//func,{data=stream}
-			lua_setmetatable(L,-2);//func,{stream}
-			lua_call(L,1,0);//
+			//stream_print(stream);
+			//printf("Created stream and everything\n");
+
+			lua_pushvalue(L,-4);//{net},{sockets},{socket},errfunc(),socket.receive(),{socket},{}
+			lua_newtable(L);//{net},{sockets},{socket},errfunc(),socket.receive(),{socket},{}
+			lua_pushlightuserdata(L,stream);//{net},{sockets},{socket},errfunc(),socket.receive(),{socket},{},ud_stream
+			lua_setfield(L,-2,"data");//{net},{sockets},{socket},errfunc(),socket.receive(),{socket},{data=stream}
+			luaL_getmetatable(L,"net.stream");//{net},{sockets},{socket},errfunc(),socket.receive(),{socket},{data=stream},{net.stream}
+			lua_setmetatable(L,-2);//{net},{sockets},{socket},errfunc(),socket.receive(),{socket},{stream}
+			//printf("About to call receive function\n");
+			lua_pcall(L,2,0,-4);//{net},{sockets},{socket},errfunc()
+			lua_pop(L,1);
+			//printf("Finished calling receive function\n");
 			//printf("Finished calling gameloop, buf is %p, size is %zu\n",buf,size);
 			nng_msg_free(msgp);
-			printf("called nn_freemsg\n");
+			//nng_free(buf,size);
+			//printf("called nn_freemsg\n");
 			free(stream);//We manually set stream->data so free_stream would crash here
-			printf("Called free on stream\n");
+			//printf("Called free on stream\n");
 		}
+
 	}
+	//printf("There are %d items left on the lua stack\n",lua_gettop(L));
+	lua_pop(L,2);
+	//for(std::map<nng_socket*,int>::iterator it = netfuncs.begin(); it != netfuncs.end(); ++it){
+		////printf("In for socket %p\n",(void*)(it->first));
+		////char* buf = NULL;
+		////size_t size;
+		////int err = nng_recv(*(it->first), &buf, &size, NNG_FLAG_NONBLOCK | NNG_FLAG_ALLOC);
+		//nng_msg *msgp;
+		////printf("About to recvmsg()\n");
+		//int err = nng_recvmsg(*(it->first), &msgp, NNG_FLAG_NONBLOCK);
+		////printf("Done recvmsg()\n");
+		////printf("Got bytes: %d EAGAIN:%d\n",bytes,EAGAIN);
+		//if(err != 0 && err != NNG_EAGAIN){
+			//printf("Net error: %s\n\t err: %d\n\teagain:%d\n",nng_strerror(err),err,NNG_EAGAIN);
+			////lua_pushstring(L,"Failed to receive");
+			////lua_error(L);
+		//}else if(err == NNG_EAGAIN){
+			////printf("EAGAIN\n");
+			////do nothing
+		//}else{
+			//printf("Calling function with stream\n");
+			//char* buf = (char*)nng_msg_body(msgp);
+			//printf("Got msg body\n");
+			//size_t size = nng_msg_len(msgp);
+			//printf("Got msg size\n");
+			////find how long until the first null character
+			//struct stream* stream = stream_create();
+			//stream->length = size;
+			//stream->data = buf;
+			//stream_print(stream);
+			//int f = it->second;
+			//if(f == -1){//not set yet
+				//lua_pushstring(L,"Got message for a socket without a receiving function");
+				//lua_error(L);
+			//}
+			//lua_rawgeti(L,LUA_REGISTRYINDEX,f);//function(stream)
+			//lua_newtable(L);//function(stream),{}
+			//lua_pushlightuserdata(L,stream);//func,{},ud_stream
+			//lua_setfield(L,-2,"data");//func,{data=stream}
+			//luaL_getmetatable(L,"net.stream");//func,{data=stream}
+			//lua_setmetatable(L,-2);//func,{stream}
+			//lua_call(L,1,0);//
+			////printf("Finished calling gameloop, buf is %p, size is %zu\n",buf,size);
+			//nng_msg_free(msgp);
+			//printf("called nn_freemsg\n");
+			//free(stream);//We manually set stream->data so free_stream would crash here
+			//printf("Called free on stream\n");
+		//}
+	//}
+	//printf("Done with net game loop\n");
 }
 
 /***
@@ -348,7 +498,7 @@ int bindsocket(lua_State*L){
 	}
 	lua_pushlightuserdata(L,fd);//{self},endpoint
 	lua_setfield(L,-2,"endpoint");//{self}
-
+	printf("Done binding socket\n");
 	return 0;
 }
 
@@ -415,27 +565,51 @@ int send(lua_State* L){
 */
 //LUA:
 //socket:receive(function(stream))
-int netreceive(lua_State* L){
-	int func = luaL_ref(L,LUA_REGISTRYINDEX);//{socket}
-	lua_getfield(L,-1,"fd");//{socket},fd
-	nng_socket *fd = (nng_socket*)lua_touserdata(L,-1);//{socket},fd
-	lua_pop(L,2);//
-	netfuncs[fd] = func;//
-	return 0;
-}
+//int netreceive(lua_State* L){
+	//if(lua_type(L,-2) != LUA_TTABLE){
+		//lua_pushstring(L,"Expected argument #1 to socket.receive to be socket");
+		//lua_error(L);
+	//}
+	//printf("two\n");
+	//if(lua_type(L,-1) != LUA_TFUNCTION){
+		//lua_pushstring(L,"Expected argument #2 to socket.receive to be function");
+		//lua_error(L);
+	//}
+	//lua_getfield(L,-1,"fd");//{socket},fd
+	//nng_socket *fd = (nng_socket*)lua_touserdata(L,-1);//{socket},fd
+	//lua_pop(L,1);//{socket}
+	//lua_getfield(L,-1,"n");//{socket},type
+	//int sockettype = lua_tonumber(L,-1);
+	//lua_pop(L,2);//
+	//switch(sockettype
+	//netfuncs[fd] = func;//
+	//nettypes[fd] = sockettype;
+	//return 0;
+//}
 
 /***
 @function socket:close()
 Closes the socket
 */
-//socket:close()
-int netclose(lua_State* L){
-	lua_getfield(L,-1,"fd");
-	nng_socket *fd = (nng_socket*)lua_touserdata(L,-1);
-	lua_pop(L,2);
+//socket:close() :: nil
+int netclose(lua_State* L){//{socket}
+	lua_getfield(L,-1,"fd");//{socket},fd
+	nng_socket *fd = (nng_socket*)lua_touserdata(L,-1);//{socket},fd
+	lua_pop(L,1);//{socket}
 	nng_close(*fd);
-	netfuncs.erase(fd);
 	free(fd);
+
+	lua_getglobal(L,"net");//{socket},{net}
+	lua_getfield(L,-1,"sockets");//{socket},{net},{sockets}
+	lua_pushvalue(L,-3);//{socket},{net},{sockets},{socket}
+	lua_pushnil(L);//{socket},{net},{sockets},{socket},nil
+	lua_settable(L,-3);//{socket},{net},{sockets}
+	lua_pop(L,2);//{socket}
+
+	lua_pushboolean(L,1);//{socket},true
+	lua_setfield(L,-2,"closed");//{socket}
+	lua_pop(L,1);
+
 	return 0;
 }
 
@@ -460,20 +634,20 @@ int socketFactory(lua_State*L){
 
 	int type = lua_tonumber(L,-1);
 	lua_pop(L,1);
-	printf("Creating a socket with type: %d, NNG_PAIR:%d\n",type,PAIR);
+	printf("Creating a socket with type: %d, NNG_PAIR:%d, NNG_BUS:%d\n",type,PAIR,BUS);
 	nng_socket *socket = (nng_socket*) malloc(sizeof(nng_socket));
 	int err;
 	switch(type){
-		case PAIR: err = nng_pair1_open(socket); break;
-		case BUS : err = nng_bus0_open(socket); break;
-		case PUB : err = nng_pub0_open(socket); break;
-		case SUB : err = nng_sub0_open(socket); break;
-		case PULL: err = nng_pull0_open(socket); break;
-		case PUSH: err = nng_push0_open(socket); break;
-		case REQ : err = nng_req0_open(socket); break;
-		case REP : err = nng_rep0_open(socket); break;
-		case SURVEY : err = nng_surveyor0_open(socket); break;
-		case RESPOND : err = nng_respondent0_open(socket); break;
+		case PAIR: err = nng_pair_open(socket); break;
+		case BUS : err = nng_bus_open(socket); break;
+		case PUB : err = nng_pub_open(socket); break;
+		case SUB : err = nng_sub_open(socket); break;
+		case PULL: err = nng_pull_open(socket); break;
+		case PUSH: err = nng_push_open(socket); break;
+		case REQ : err = nng_req_open(socket); break;
+		case REP : err = nng_rep_open(socket); break;
+		case SURVEY : err = nng_surveyor_open(socket); break;
+		case RESPOND : err = nng_respondent_open(socket); break;
 		default:
 			printf("Unknown socket type: %d",type);
 			break;
@@ -489,9 +663,20 @@ int socketFactory(lua_State*L){
 	lua_newtable(L);//{}
 	lua_pushlightuserdata(L,socket);//{},c
 	lua_setfield(L,-2,"fd");//{fd=c}
+	lua_pushnumber(L,type);//{fd=c},type
+	lua_setfield(L,-2,"n");//{fd=c,n=type}
 
-	luaL_getmetatable(L,"net.pair_socket");//{},{m_net.pair_socket}
+	printf("Metatable set\n");
+	luaL_getmetatable(L,"net.pair_socket");//{fd=c,n=type},{m_net.pair_socket}
 	lua_setmetatable(L,-2);//{socket}
+
+	lua_getglobal(L,"net");//{socket},{net}
+	lua_getfield(L,-1,"sockets");//{socket},{net},{sockets}
+	lua_pushvalue(L,-3);//{socket},{net},{sockets},{socket}
+	lua_pushboolean(L,1);//{socket},{net},{sockets},{socket},true
+	lua_settable(L,-3);//{socket},{net},{sockets}
+	lua_pop(L,2);//{socket}
+
 	printf("Finished making the socket, returning\n");
 	return 1;
 }
@@ -516,7 +701,8 @@ static const struct luaL_Reg pair_socket_m[] = {
 	{"bind", bindsocket},
 	{"connect", connectsocket},
 	{"send",send},
-	{"receive", netreceive},
+	{"block_recv",block_recv},
+	//{"receive", netreceive},
 	{"close",netclose},
 	{NULL,NULL}
 };
@@ -538,6 +724,8 @@ void loadNetLibs(lua_State* L){
 
 	//A table to hold all our net funcs
 	lua_newtable(L);//{}
+	lua_newtable(L);//{},{}
+	lua_setfield(L,-2,"sockets");//{sockets = {}}
 
 	//Push some enums
 	//Protocols