lihao/skynet
1
0
Fork 0
mirror of https://github.com/cloudwu/skynet.git synced 2026-07-24 03:53:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

Support message redirect

Browse Source
This commit is contained in:
云风
2012-08-09 21:50:33 +08:00
parent d3f92b32d5
commit 24790ca5e4
9 changed files with 101 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
connection/main.c
View File

@@ -65,12 +65,12 @@ _connect(struct connection_server *server, const char * ipaddr, size_t sz, int s
tmp[sz] = '\0';
int id = connection_open(server->pool, ipaddr);
if (id == 0) {
skynet_send(server->ctx, reply, session, NULL, 0, 0);
skynet_send(server->ctx, NULL, reply, session, NULL, 0, 0);
return;
}
char idstring[20];
int n = sprintf(idstring, "%d", id);
skynet_send(server->ctx, reply, session, idstring, n, 0);
skynet_send(server->ctx, NULL, reply, session, idstring, n, 0);
}
static void
@@ -114,7 +114,7 @@ _read(struct connection_server *server, const char * param, size_t sz, int sessi
if (buffer == NULL) {
int id = connection_id(server->pool, handle);
if (id == 0) {
skynet_send(server->ctx, reply, session, NULL, 0, 0);
skynet_send(server->ctx, NULL, reply, session, NULL, 0, 0);
return;
}
--id;
@@ -128,7 +128,7 @@ _read(struct connection_server *server, const char * param, size_t sz, int sessi
return;
}
} else {
skynet_send(server->ctx, reply, session, buffer, size, 0);
skynet_send(server->ctx, NULL, reply, session, buffer, size, 0);
}
}
@@ -155,7 +155,7 @@ _readline(struct connection_server *server, const char * param, size_t sz, int s
if (buffer == NULL) {
int id = connection_id(server->pool, handle);
if (id == 0) {
skynet_send(server->ctx, reply, session, NULL, 0, 0);
skynet_send(server->ctx, NULL, reply, session, NULL, 0, 0);
return;
}
--id;
@@ -169,7 +169,7 @@ _readline(struct connection_server *server, const char * param, size_t sz, int s
return;
}
} else {
skynet_send(server->ctx, reply, session, buffer, sz, 0);
skynet_send(server->ctx, NULL, reply, session, buffer, sz, 0);
}
}
@@ -199,7 +199,7 @@ _poll(struct connection_server *server) {
struct reply * r = &server->reply[id];
addr[0] = ':';
_id_to_hex(addr+1, id);
skynet_send(server->ctx, addr , r->session, NULL, 0, 0);
skynet_send(server->ctx, NULL, addr , r->session, NULL, 0, 0);
} else {
assert(server->poll >= 0);
if (server->poll > 0) {
@@ -215,7 +215,7 @@ _poll(struct connection_server *server) {
struct reply * r = &server->reply[id];
addr[0] = ':';
_id_to_hex(addr+1, r->dest);
skynet_send(server->ctx, addr, r->session, buffer, sz, 0);
skynet_send(server->ctx, NULL, addr, r->session, buffer, sz, 0);
}
}
}

31
gate/main.c
View File

@@ -12,6 +12,7 @@
struct connection {
char * agent;
char * client;
int connection_id;
int uid;
};
@@ -55,12 +56,16 @@ _parm(char *msg, int sz, int command_sz) {
}
static void
_forward_agent(struct gate * g, int id, char * addr) {
_forward_agent(struct gate * g, int id, const char * agentaddr, const char *clientaddr) {
struct connection * agent = _id_to_agent(g,id);
if (agent->agent) {
free(agent->agent);
}
agent->agent = strdup(addr);
agent->agent = strdup(agentaddr);
if (agent->client) {
free(agent->client);
}
agent->client = strdup(clientaddr);
}
static void
@@ -88,12 +93,17 @@ _ctrl(struct skynet_context * ctx, struct gate * g, const void * msg, int sz) {
}
if (memcmp(command,"forward",i)==0) {
_parm(tmp, sz, i);
char * start = tmp;
char * data = strsep(&start, " ");
int id = strtol(data , NULL, 10);
if (start) {
_forward_agent(g, id, start);
char * client = tmp;
char * idstr = strsep(&client, " ");
if (client == NULL) {
return;
}
int id = strtol(idstr , NULL, 10);
char * agent = strsep(&client, " ");
if (client == NULL) {
return;
}
_forward_agent(g, id, agent, client);
return;
}
skynet_error(ctx, "[gate] Unkown command : %s", command);
@@ -107,19 +117,20 @@ _report(struct gate *g, struct skynet_context * ctx, const char * data, ...) {
int n = vsnprintf(tmp, sizeof(tmp), data, ap);
va_end(ap);
skynet_send(ctx, g->watchdog, 0, tmp, n, 0);
skynet_send(ctx, NULL, g->watchdog, 0, tmp, n, 0);
}
static void
_forward(struct skynet_context * ctx,struct gate *g, int uid, void * data, size_t len) {
struct connection * agent = _id_to_agent(g,uid);
if (agent->agent) {
skynet_send(ctx, agent->agent, 0, data, len, 0);
// todo: client package has not session , send 0
skynet_send(ctx, agent->client, agent->agent, 0, data, len, 0);
} else {
char * tmp = malloc(len + 32);
int n = snprintf(tmp,len+32,"%d data ",uid);
memcpy(tmp+n,data,len);
skynet_send(ctx, g->watchdog, 0, tmp, len + n, DONTCOPY);
skynet_send(ctx, NULL, g->watchdog, 0, tmp, len + n, DONTCOPY);
}
}

37
lualib-src/lua-skynet.c
View File

@@ -110,20 +110,20 @@ _send(lua_State *L) {
++index;
}
if (lua_gettop(L) == index + 1) {
session = skynet_send(context, dest, session , NULL, 0, 0);
session = skynet_send(context, NULL, dest, session , NULL, 0, 0);
} else {
int type = lua_type(L,index+2);
if (type == LUA_TSTRING) {
size_t len = 0;
void * msg = (void *)lua_tolstring(L,index+2,&len);
session = skynet_send(context, dest, session , msg, len, 0);
session = skynet_send(context, NULL, dest, session , msg, len, 0);
} else if (type == LUA_TNIL) {
session = skynet_send(context, dest, session , NULL, 0, 0);
session = skynet_send(context, NULL, dest, session , NULL, 0, 0);
} else {
luaL_checktype(L,index+2, LUA_TLIGHTUSERDATA);
void * msg = lua_touserdata(L,index+2);
int size = luaL_checkinteger(L,index+3);
session = skynet_send(context, dest, session, msg, size, DONTCOPY);
session = skynet_send(context, NULL, dest, session, msg, size, DONTCOPY);
}
}
if (session < 0) {
@@ -133,6 +133,34 @@ _send(lua_State *L) {
return 1;
}
static int
_redirect(lua_State *L) {
struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
const char * dest = luaL_checkstring(L,1);
const char * source = luaL_checkstring(L,2);
int session = luaL_checkinteger(L,3);
if (lua_gettop(L) == 3) {
session = skynet_send(context, source, dest, session , NULL, 0, 0);
} else {
int type = lua_type(L,4);
if (type == LUA_TSTRING) {
size_t len = 0;
void * msg = (void *)lua_tolstring(L,4,&len);
skynet_send(context, source, dest, session , msg, len, 0);
} else if (type == LUA_TNIL) {
session = skynet_send(context, source, dest, session , NULL, 0, 0);
} else {
luaL_checktype(L, 4, LUA_TLIGHTUSERDATA);
void * msg = lua_touserdata(L,4);
int size = luaL_checkinteger(L,5);
skynet_send(context, source, dest, session, msg, size, DONTCOPY);
}
}
return 0;
}
static int
_error(lua_State *L) {
struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
@@ -156,6 +184,7 @@ luaopen_skynet_c(lua_State *L) {
luaL_checkversion(L);
luaL_Reg l[] = {
{ "send" , _send },
{ "redirect", _redirect },
{ "command" , _command },
{ "callback" , _callback },
{ "error", _error },

9
lualib/skynet.lua
View File

@@ -81,10 +81,11 @@ function skynet.setenv(key, value)
c.command("SETENV",key .. " " ..value)
end
skynet.send = c.send
skynet.pack = c.pack
skynet.tostring = c.tostring
skynet.unpack = c.unpack
skynet.send = assert(c.send)
skynet.redirect = assert(c.redirect)
skynet.pack = assert(c.pack)
skynet.tostring = assert(c.tostring)
skynet.unpack = assert(c.unpack)
function skynet.call(addr, deseri , ...)
local t = type(deseri)

4
service-src/service_broker.c
View File

@@ -63,7 +63,7 @@ _cb(struct skynet_context * context, void * ud, int session, const char * addr,
_init(b, session, msg, sz);
if (b->init == DEFAULT_NUMBER) {
skynet_command(context, "REG", b->name);
skynet_send(context, LAUNCHER, 0, NULL, 0, 0);
skynet_send(context, NULL, LAUNCHER, 0, NULL, 0, 0);
}
} else {
_forward(b, context);
@@ -90,7 +90,7 @@ broker_init(struct broker *b, struct skynet_context *ctx, const char * args) {
if (len == 0)
return 1;
for (i=0;i<DEFAULT_NUMBER;i++) {
int id = skynet_send(ctx, LAUNCHER , -1, service , len, 0);
int id = skynet_send(ctx, NULL, LAUNCHER , -1, service , len, 0);
assert(id > 0 && id <= DEFAULT_NUMBER);
}

15
service/agent.lua
View File

@@ -1,17 +1,8 @@
local skynet = require "skynet"
local client = ...
skynet.dispatch(function(msg, sz , session, address)
local message = skynet.tostring(msg,sz)
if session == 0 then
print("client command",message)
local result = skynet.call("SIMPLEDB",message)
skynet.send(client, result)
else
print("server command",message)
if msg == "CLOSE" then
skynet.kill(client)
skynet.exit()
end
end
print("command source",address)
local result = skynet.call("SIMPLEDB",message)
skynet.send(address, result)
end)

28
service/watchdog.lua
View File

@@ -4,41 +4,45 @@ local command = {}
local agent_all = {}
local gate = skynet.launch("gate" , skynet.self(), ...)
print("gate",gate)
function command:open(parm)
local fd,addr = string.match(parm,"(%d+) ([^%s]+)")
fd = tonumber(fd)
skynet.send("LOG", string.format("%d %d %s",self,fd,addr))
print("agent open",self,string.format("%d %d %s",self,fd,addr))
local agent = skynet.launch("snlua","agent")
local client = skynet.launch("client",fd)
skynet.send("LOG", "client " .. client)
local agent = skynet.launch("snlua","agent",client)
print("watchdog launch agent client:",agent,client)
if agent then
agent_all[self] = agent
skynet.send(gate, "forward ".. self .. " " .. agent)
agent_all[self] = { agent , client }
skynet.send(gate, "forward ".. self .. " " .. agent .. " " .. client)
end
end
function command:close()
skynet.send("LOG", string.format("close %d",self))
skynet.send(agent_all[self],-1,"CLOSE")
print("agent close",self,string.format("close %d",self))
local agent = agent_all[self]
agent_all[self] = nil
skynet.kill(agent[1])
skynet.kill(agent[2])
end
function command:data(data)
function command:data(data, session)
local agent = agent_all[self]
if agent then
skynet.send(agent, data)
skynet.redirect(agent[1], agent[2], session, data)
else
skynet.send("LOG", string.format("data %d size=%d",self,#data))
skynet.error(string.format("agent data drop %d size=%d",self,#data))
end
end
skynet.dispatch(function(msg, sz)
skynet.dispatch(function(msg, sz, session, address)
local message = skynet.tostring(msg,sz)
local id, cmd , parm = string.match(message, "(%d+) (%w+) ?(.*)")
id = tonumber(id)
local f = command[cmd]
if f then
f(id,parm)
f(id,parm,session)
else
skynet.error(string.format("[watchdog] Unknown command : %s",message))
end

2
skynet-src/skynet.h
View File

@@ -10,7 +10,7 @@ struct skynet_context;
void skynet_error(struct skynet_context * context, const char *msg, ...);
const char * skynet_command(struct skynet_context * context, const char * cmd , const char * parm);
int skynet_send(struct skynet_context * context, const char * addr , int session, void * msg, size_t sz, int flags);
int skynet_send(struct skynet_context * context, const char * source, const char * addr , int session, void * msg, size_t sz, int flags);
void skynet_forward(struct skynet_context *, const char * addr);
typedef void (*skynet_cb)(struct skynet_context * context, void *ud, int session, const char * addr , const void * msg, size_t sz);

16
skynet-src/skynet_server.c
View File

@@ -378,7 +378,15 @@ skynet_forward(struct skynet_context * context, const char * addr) {
}
int
skynet_send(struct skynet_context * context, const char * addr , int session, void * data, size_t sz, int flags) {
skynet_send(struct skynet_context * context, const char * source, const char * addr , int session, void * data, size_t sz, int flags) {
uint32_t source_handle;
if (source == NULL) {
source_handle = context->handle;
} else {
assert (source[0] == ':');
source_handle = strtoul(source+1, NULL, 16);
}
char * msg;
if ((flags & DONTCOPY) || data == NULL) {
msg = data;
@@ -394,7 +402,7 @@ skynet_send(struct skynet_context * context, const char * addr , int session, vo
}
uint32_t des = 0;
if (addr[0] == ':') {
des = strtol(addr+1, NULL, 16);
des = strtoul(addr+1, NULL, 16);
} else if (addr[0] == '.') {
des = skynet_handle_findname(addr + 1);
if (des == 0) {
@@ -404,7 +412,7 @@ skynet_send(struct skynet_context * context, const char * addr , int session, vo
}
} else {
struct skynet_message smsg;
smsg.source = context->handle;
smsg.source = source_handle;
smsg.session = session_id;
smsg.data = msg;
smsg.sz = sz;
@@ -415,7 +423,7 @@ skynet_send(struct skynet_context * context, const char * addr , int session, vo
assert(des > 0);
struct skynet_message smsg;
smsg.source = context->handle;
smsg.source = source_handle;
smsg.session = session_id;
smsg.data = msg;
smsg.sz = sz;