Skip to content
Snippets Groups Projects
Commit 22ed878a authored by mcmlxxix's avatar mcmlxxix
Browse files

Added file synchronization functions. 

Changed just about everything. See diff.
parent 632229ec
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
Showing
with 138 additions and 69 deletions
exec/commservice.js
+ 138
− 69
View file @ 22ed878a
//load("commsync.js");
/*
Inter-BBS/Inter-Node socket service
for Synchronet v3.15+
......@@ -30,24 +29,32 @@
*/
const normal_scope= "#";
const global_scope= "!";
const node_prefix= "*";
const local_prefix= "&";
//Data type constants
const normal= "#";
const global= "!";
const priv= "~";
const query= "?";
//Connection type constants
const remote= "*";
const local= "&";
const filesync= "@";
const hub_address=argv[0];
const hub_port=argv[1];
const connection_timeout=2;
const connection_attempts=5;
var local_socks=[];
var remote_socks=[];
var local_sessions=[];
var remote_sessions=[];
var data_queue=[];
var remote_hub=false;
var background=[];
var hub=false;
//main service loop
while(!server.terminated)
{
if(server.terminated) break;
sock_cycle();
if(server.socket.poll()<1)
{
......@@ -63,7 +70,7 @@ function sock_cycle()
{
if(hub_address && hub_port)
{
if(!remote_hub || !remote_hub.is_connected)
if(!hub || !hub.is_connected)
hub_connect();
}
else if(hub_address || hub_port)
......@@ -72,10 +79,11 @@ function sock_cycle()
}
inbound();
outbound();
send_updates();
}
else
{
if(remote_hub && remote_hub.is_connected)
if(hub && hub.is_connected)
{
hub_disconnect();
}
......@@ -84,16 +92,16 @@ function sock_cycle()
function inbound()
{
//read data from each socket
for(var l in local_socks)
for(var l in local_sessions)
{
for(var s=0;s<local_socks[l].length;s++)
for(var s=0;s<local_sessions[l].length;s++)
{
var socket=local_socks[l][s].sock;
if(socket.is_connected)
var session=local_sessions[l][s];
if(session.sock.is_connected)
{
if(socket.data_waiting)
if(session.sock.data_waiting)
{
socket_receive(socket);
socket_receive(session.sock);
}
}
else
......@@ -102,21 +110,21 @@ function inbound()
}
}
}
for(var r=0;r<remote_socks.length;r++)
for(var r=0;r<remote_sessions.length;r++)
{
var node=remote_socks[r];
if(node.sock.is_connected)
var session=remote_sessions[r];
if(session.sock.is_connected)
{
socket_receive(node.sock);
socket_receive(session.sock);
}
else
{
r=delete_remote_session(r);
}
}
if(remote_hub && remote_hub.is_connected)
if(hub && hub.is_connected)
{
socket_receive(remote_hub);
socket_receive(hub);
}
}
function outbound()
......@@ -126,14 +134,14 @@ function outbound()
{
var data=data_queue[d];
//if the central hub is connected and was not the origin of the data, send thru
if(remote_hub && remote_hub.is_connected)
if(hub && hub.is_connected)
{
socket_send_remote(data,remote_hub);
socket_send_remote(data,hub);
}
//send data to all remotely connected nodes
for(var r=0;r<remote_socks.length;r++)
for(var r=0;r<remote_sessions.length;r++)
{
var socket=remote_socks[r].sock;
var socket=remote_sessions[r].sock;
if(socket.is_connected)
{
socket_send_remote(data,socket);
......@@ -146,12 +154,12 @@ function outbound()
switch(data.scope)
{
//if data is meant to be sent to all connected clients, do so (usually chat messages)
case global_scope:
for(var l in local_socks)
case global:
for(var l in local_sessions)
{
for(var s=0;s<local_socks[l].length;s++)
for(var s=0;s<local_sessions[l].length;s++)
{
var socket=local_socks[l][s].sock;
var socket=local_sessions[l][s].sock;
if(socket.is_connected)
{
socket_send_local(data,socket);
......@@ -164,11 +172,11 @@ function outbound()
}
break;
//distribute data to appropriate local sessions
case normal_scope:
case normal:
default:
for(var s=0;local_socks[data.session] && s<local_socks[data.session].length;s++)
for(var s=0;local_sessions[data.session] && s<local_sessions[data.session].length;s++)
{
var socket=local_socks[data.session][s].sock;
var socket=local_sessions[data.session][s].sock;
if(socket.is_connected)
{
socket_send_local(data,socket);
......@@ -186,7 +194,7 @@ function outbound()
function delete_local_session(session_id,index)
{
log("local socket connection terminated: " + session_id);
local_socks[session_id].splice(index,1);
local_sessions[session_id].splice(index,1);
return index-1;
}
function socket_receive(socket)
......@@ -194,12 +202,24 @@ function socket_receive(socket)
if(socket.data_waiting)
{
//store data in master array to be distributed later
var raw=socket.recvline(4092,connection_timeout);
var scope=raw.charAt(0);
var session_id=raw.substring(1,raw.indexOf(":"));
var data=raw.substr(raw.indexOf(":")+1);
var raw=socket.recvline(16384,connection_timeout);
var data=raw.split("");
var scope=data[0].charAt(0);
var session=data[0].substr(1);
data_queue.push(new Packet(scope,session_id,socket.descriptor,data));
switch(scope)
{
case global:
case normal:
data_queue.push(new Packet(scope,socket.descriptor,session,data[1]));
break;
case query:
handle_query(socket,data);
break;
default:
log("received unknown data type");
break;
}
}
}
function socket_send_local(data,socket)
......@@ -214,61 +234,65 @@ function socket_send_remote(data,socket)
{
if(socket.descriptor!=data.descriptor)
{
var d=data.scope + data.session + ":" + data.data + "\r\n";
var d=data.scope + data.session + "" + data.data + "\r\n";
socket.write(d);
}
}
function delete_remote_session(index)
{
log("remote socket connection terminated: " + remote_socks[index].sock.remote_ip_address);
remote_socks.splice(index,1);
log("remote socket connection terminated: " + remote_sessions[index].sock.remote_ip_address);
remote_sessions.splice(index,1);
return index-1;
}
function hub_connect()
{
remote_hub=new Socket();
remote_hub.bind(0,server.interface_ip_address);
log("connecting to hub");
hub=new Socket();
hub.bind(0,server.interface_ip_address);
//if a central hub address is provided, attempt connection
for(var t=0;t<connection_attempts;t++)
{
if(remote_hub.connect(hub_address,hub_port,connection_timeout))
if(hub.connect(hub_address,hub_port,connection_timeout))
{
//send node identifier and bbs name so hub knows this is a distribution point
remote_hub.send(node_prefix + system.name + "\r\n");
hub.send(remote + system.name + "\r\n");
log("connection to " + hub_address + " successful");
return true;
}
mswait(50);
}
log("connection to " + hub_address + " failed with error " + remote_hub.error);
log("connection to " + hub_address + " failed with error " + hub.error);
return false;
}
function hub_disconnect()
{
log("disconnecting from main hub: " + hub_address);
remote_hub.close();
hub.close();
}
function store_socket(sock)
{
//receive connection identifier from incoming socket connection (should always be first transmission)
var handshake=sock.recvline(4092,connection_timeout);
var handshake=sock.recvline(1024,connection_timeout);
var identifier=handshake.charAt(0);
var session_id=handshake.substr(1);
switch(identifier)
{
case node_prefix:
remote_socks.push(new Node(sock,session_id));
log("remote node connection from: " + session_id + "@" + sock.remote_ip_address);
case remote:
log("remote connection from: " + sock.remote_ip_address + ":" + session_id);
remote_sessions.push(new Session(sock,session_id));
break;
case local:
log("local connection from: " + sock.remote_ip_address + ":" + session_id);
if(!local_sessions[session_id]) local_sessions[session_id]=[];
local_sessions[session_id].push(new Session(sock,session_id));
break;
case local_prefix:
if(!local_socks[session_id]) local_socks[session_id]=[];
local_socks[session_id].push(new Session(sock));
log("local node connection: " + session_id);
case filesync:
log("file sync connection from: " + sock.remote_ip_address + ":" + session_id);
distribute_files(sock,handshake);
break;
default:
log("unknown connection type");
log("unknown connection type: " + identifier);
sock.close();
break;
}
......@@ -276,18 +300,68 @@ function store_socket(sock)
function count_local_sockets()
{
var count=0;
for(var l in local_socks)
for(var l in local_sessions)
{
if(local_socks[l].length) count+=local_socks[l].length;
if(local_sessions[l].length) count+=local_sessions[l].length;
}
return count;
}
function count_remote_sockets()
{
return remote_socks.length;
return remote_sessions.length;
}
function handle_query(socket,query)
{
var data=query.split("");
var session_id=data[0].substr(1);
var descriptor=data[1];
var task= data[2];
switch(task)
{
case "#who":
whos_online(socket);
break;
case "#finduser":
find_user(socket,data);
break;
case "#pageuser":
page_user(socket,data);
break;
default:
log("unknown query type: " + task);
break;
}
}
function distribute_files(socket,query)
{
background.push(load(true,"filesync.js",socket.descriptor,query,hub_address,hub_port));
}
function send_updates(socket)
{
while(background.length)
{
var bgq=background.shift();
while(bgq.data_waiting)
{
var data=bgq.read();
for(var r=0;r<remote_sessions.length;r++)
{
var sock=remote_sessions[r].sock;
if(data.remote_ip_address != socket.remote_ip_address)
{
var query=("@" + data.session_id + "#send" + data.filename + "" + data.filedate);
load(true,"filesync.js",sock.descriptor,query);
}
}
}
}
}
//TODO
function whos_online()
{
}
function find_user()
{
}
......@@ -295,20 +369,15 @@ function page_user()
{
}
function Packet(scope,session,descriptor,data)
{
function Packet(scope,descriptor,session,data)
{
this.session=session;
this.descriptor=descriptor;
this.data=data;
this.scope=scope;
}
function Node(socket,name)
{
this.sock=socket;
this.name=name;
}
function Session(socket,session)
{
this.sock=socket;
this.data=[];
this.session=session;
}
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment