OS Instances Connector [backup] [1035030]

!7

static int updateInterval = 10;
static long timeout = 30000+10000; // idle time + latency
static int port = 6000;
static int maxLineLength = 10*1024*1024; // change #1020500 too if you change this

import java.nio.*;
import java.nio.channels.*;

static ConnectedInstances cI;
static ReliableSingleThread rst = new(r update);
static int theNumber;
static Map<SocketChannel, State> dataMapper = synchroMap();
static ServerSocketChannel serverChannel;
static Selector selector;
static Map<S, Channel> channels = synchroMap();
static Lock channelsLock = lock();
static java.util.Timer mainTimer;
sbool debug;

sclass State {
  S socketID = aGlobalID(), computerID;
  int countSeen = -1;
  new LineBuffer lineBuffer;
  long lastMessage = sysNow();
  Map<S, Channel> subbedChannels = synchroLinkedHashMap();
  Lock lock = lock();
  
  // TODO: clear eventually to avoid stalling attacks
  L<byte[]> toSend = synchroLinkedList();
}

sclass Channel<A> {
  S name;
  Set<SocketChannel> sockets = synchroSet();
  Map<SocketChannel, A> dataMapper = synchroMap();
  Lock lock = lock();
  
  void addSocket(SocketChannel socket) {
    lock lock;
    sockets.add(socket);
    initialMessage(socket);
  }
  
  void removeSocket(SocketChannel socket) {
    lock lock;
    sockets.remove(socket);
  }
  
  void update() {}
  void initialMessage(SocketChannel socket) {}
  void handleLine(S line, SocketChannel socket) {}
}

sclass ComputerIDsChannel extends Channel<O> {
  L<S> value;
  
  void initialMessage(SocketChannel socket) {
    lock lock;
    sendTo(socket);
  }
  
  void update() {
    if (set_trueIfChanged(this, value := asList(cI.set())))
      for (SocketChannel s : sockets) sendTo(s);
  }
  
  void sendTo(SocketChannel socket) {
    sendLineToSocket(socket, "ComputerIDs = " + struct(value));
  }
}

sclass ComputerCountChannel extends Channel<O> {
  int value;
  
  void initialMessage(SocketChannel socket) {
    lock lock;
    sendTo(socket);
  }
  
  void update() {
    if (set_trueIfChanged(this, value := cI.value()))
      for (SocketChannel s : sockets) sendTo(s);
  }
  
  void sendTo(SocketChannel socket) {
    sendLineToSocket(socket, "ComputerCount = " + value);
  }
}

sclass PublicChatChannel extends Channel<O> {
  L<S> msgs = synchroLinkedList();
  
  void handleLine(S line, SocketChannel socket) {
    postMessage(unquote(line));
  }
  
  void postMessage(S msg) {
    msgs.add(msg);
  }
  
  void update {
    S msg;
    while ((msg = syncPopFirst(msgs)) != null) {
      S quoted = quote(msg);
      //print("Chat send: " + shorten(quoted, 20));
      quoted = name + ":" + quoted;
      for (SocketChannel s : sockets)
        sendLineToSocket(s, quoted);
    }
  }
}

sclass PrivateChatChannel extends Channel<O> {
  S prefix = "";
  new MultiMap<S, SocketChannel> clients; // key: computer ID

  void initialMessage(SocketChannel socket) {
    clients.put(computerID(socket), socket);
  }
  
  void removeSocket(SocketChannel socket) {
    super.removeSocket(socket);
    clients.remove(computerID(socket), socket);
  }
  
  void handleLine(S line, SocketChannel socket) {
    pcall {
      int i = indexOf(line, ' ');
      S computerID = takeFirst(line, i);
      while licensed {
        SocketChannel socket2 = clients.getFirst(computerID);
        if (socket2 == null) break;
        try {
          S sender = computerID(socket);
          if (l(line) >= 10000)
            print("Forwarding line from " + sender + " to " + computerID + ": " + l(line));
          sendLineToSocket(socket2, prefix + "privately from " + sender + ": " + substring(line, i+1));
          break;
        } catch e {
          closeSocket(socket2);
          clients.remove(computerID, socket2);
        }
      }
    }
  }
}

sclass SnippetUpdatesChannel extends Channel<O> {
  ManualTailFile tail;
  
  *() {
    tail = manualTailFile_newOnly(programFile(#1019175, "changes.log"),
      vfAppendToLineBuffer(unquotingLineBuffer(voidfunc(S line) { sendLine(line) })));
  }
  
  void update {
    tail.check();
  }
  
  void sendLine(S line) {
    print("SnippetUpdatesChannel send: " + line);
    line = assertNoLineBreaks(line);
    print("Sockets: " + l(sockets));
    //S quoted = quote(line);
    for (SocketChannel s : sockets)
      sendLineToSocket(s, "snippetUpdates:" + line);
  }
}

static State state(SocketChannel socket) {
  ret dataMapper.get(socket);
}

sS computerID(SocketChannel socket) {
  State state = state(socket);
  ret state == null ? null : state.computerID;
}

p {
  cI = new ConnectedInstances;
  cI.setAfterglowInterval(1000);
  cI.connected2.onChange(rst);
  cI.start();
  
  openServerSocket();
  
  // restart socket every hour (and once now for testing - XX)
  /*doEvery(10.0, 60*60.0, r {
    pcall { cleanMeUp_socket(); }
    openServerSocket();
  });*/
}

svoid openServerSocket ctex {
  print("Opening server socket");
  selector = Selector.open();
  serverChannel = ServerSocketChannel.open();
  serverChannel.configureBlocking(false);
  
  InetSocketAddress sockAddr = new InetSocketAddress(port);
  serverChannel.socket().bind(sockAddr);
  print("Server socket bound");
  serverChannel.register(selector, SelectionKey.OP_ACCEPT);
  
  thread "Server Accept Loop" { acceptLoop(); }
  
  mainTimer = doEvery(updateInterval, r {
    for (SocketChannel socket : keysList(dataMapper)) {
      State state = state(socket);
      if (state != null && sysTime() >= state.lastMessage+timeout) pcall {
        closeSocket(socket);
      }
    }
    for (Channel channel : values(channels)) pcall {
      lock channel.lock;
      channel.update();
    }
  });
}

svoid cleanMeUp_socket() ctex {
  if (serverChannel != null)
    serverChannel.close();
  for (SocketChannel socket : keys(getAndClearMap(dataMapper))) pcall {
    socket.close();
  }
}

svoid acceptLoop ctex {
  while licensed {
    // wait for events
    selector.select();

    Iterator<SelectionKey> keys = selector.selectedKeys().iterator();
    while (keys.hasNext()) {
      SelectionKey key = keys.next();
      keys.remove();
      debug "have key";
      
      if (!key.isValid())
        continue with debug "key not valid";
        
      pcall {
        if (key.isAcceptable()) {
          debug "Key is acceptable";
          acceptOnSocket(key);
        } else if (key.isReadable()) {
          debug "Key is readable";
          readFromSocket(key);
        } else if (key.isWritable()) {
          debug "Key is writable";
          writeToSocket(key);
        } else
          debug "Unknown key type";
      }
    }
  }
}

//accept a connection made to this channel's socket
static void acceptOnSocket(SelectionKey key) throws IOException {
  ServerSocketChannel serverChannel = cast key.channel();
  SocketChannel channel = serverChannel.accept();
  channel.configureBlocking(false);
  Socket actualSocket = channel.socket();
  SocketAddress remoteAddr = actualSocket.getRemoteSocketAddress();
  print("Connected to: " + remoteAddr);
  
  // register channel with selector for further IO
  dataMapper.put(channel, new State);
  channel.register(selector, SelectionKey.OP_READ);
}

svoid onSocketClose(SocketChannel socket) {
  Socket actualSocket = socket.socket();
  SocketAddress remoteAddr = actualSocket.getRemoteSocketAddress();
  print("Connection closed by client: " + remoteAddr);
  State state = state(socket);
  if (state != null && state.computerID != null)
    cI.lostConnection(state.computerID);
  dataMapper.remove(socket);
  removeFromAllChannels(socket);
}

svoid closeSocket(SocketChannel socket) {
  pcall { socket.close(); }
  pcall { onSocketClose(socket); }
}

static void readFromSocket(SelectionKey key) throws IOException {
  final SocketChannel socket = cast key.channel();
  ByteBuffer buffer = ByteBuffer.allocate(1024);
  int numRead = socket.read(buffer);
  if (numRead == -1) {
    closeSocket(socket);
    key.cancel();
    ret;
  }

  S s = new String(buffer.array(), 0, numRead);
  //print("Got data from client: " + s);
  final State state = state(socket);
  if (state != null) {
    state.lastMessage = sysNow();
    state.lineBuffer.append(s, voidfunc(S line) {
      //if (nempty(line)) print("Got line: " + line);
      pcall { handleLine(line, socket, state); }
    });
    if (state.lineBuffer.size() > maxLineLength) {
      print("Bad client (line too long");
      closeSocket(socket);
    }
  }
}

static void writeToSocket(SelectionKey key) throws IOException {
  final SocketChannel socket = cast key.channel();
  socket.register(selector, SelectionKey.OP_READ);
  State state = state(socket);
  lock state.lock;
  if (empty(state.toSend)) ret;
  byte[] data = popFirst(state.toSend);
  print("Sending postponed data (" + n2(l(data), "byte") + ") to " + state.computerID);
  sendBytesToSocket(socket, data);
}

svoid sendBytesToSocket(SocketChannel socket, byte[] bytes) throws IOException {
  int length = l(bytes);
  State state = state(socket);
  lock state.lock;
  if (nempty(state.toSend)) {
    print("Postponing send to " + state.computerID + " because data already queued");
    ret with state.toSend.add(bytes);
  }
  ByteBuffer buf = ByteBuffer.wrap(bytes);
  int n = socket.write(buf);
  if (n < length) {
    byte[] remaining = subByteArray(bytes, n);
    state.toSend.add(remaining);
    print("Postponing sending " + n2(l(remaining), "byte") + " to " + state.computerID);
    socket.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);
  }
}

svoid handleLine(S line, SocketChannel socket, State state) {
  new Matches m;
  if (empty(line)) ret with sendLineToSocket(socket, ""); // idle
  
  if (jmatch("computerID=*", line, m)) {
    if (state.computerID != null || !possibleComputerID($1)) ret;
    print("Got computer ID: " + $1);
    state.computerID = $1;
    cI.gotConnection(state.computerID);
    ret;
  }
  
  if (jmatch("sub *", line, m)) {
    Channel channel = getChannelForSocket(socket, $1);
    if (channel == null)
      sendLineToSocket(socket, "Channel " + $1 + " not found");
    else
      sendLineToSocket(socket, "Subbed " + $1);
    ret;
  }
  
  L<S> tok = javaTok(line);
  if (eq(get(tok, 3), ":")) {
    Channel channel = state.subbedChannels.get(get(tok, 1));
    if (channel != null)
      channel.handleLine(joinSubList(tok, 5), socket);
  }
  
  // ignore unknown line
}

html { ret "Use port " + port; }

svoid sendLineToSocket(SocketChannel socket, S s) ctex {
  byte[] bytes = toUtf8(s + "\n");
  sendBytesToSocket(socket, bytes);
}

svoid removeFromAllChannels(SocketChannel socket) {
  lock channelsLock;
  for (S name, Channel c : cloneMap(channels)) { pcall {
    c.removeSocket(socket);
    if (empty(c.sockets)) {
      print("Deleting channel: " + name);
      channels.remove(name);
    }
  }}
}

static Channel getChannelForSocket(SocketChannel socket, S name) {
  lock channelsLock;
  Channel c = channels.get(name);
  if (c == null) {
    c = makeChannel(name);
    c.name = name;
    if (c == null) null;
    print("New channel: " + name);
    channels.put(name, c);
    lock c.lock;
    c.update();
  }
  c.addSocket(socket);
  state(socket).subbedChannels.put(name, c);
  ret c;
}

static Channel makeChannel(S name) {
  if (eq(name, "computerIDs")) ret new ComputerIDsChannel;
  if (eq(name, "computerCount")) ret new ComputerCountChannel;
  if (eq(name, "chat")) ret new PublicChatChannel;
  if (eq(name, "machineChat")) ret new PublicChatChannel;
  if (eq(name, "privateChat")) ret new PrivateChatChannel;
  if (eq(name, "privateMachineChat")) ret nu(PrivateChatChannel, prefix := name + ":");
  if (eq(name, "snippetUpdates")) ret new SnippetUpdatesChannel;
    
  ret null with print("Unknown channel: " + name);
}

svoid update {
  print("OS Instances update");
  if (empty(dataMapper)) ret;

  int value = cI.value();
  //if (value == theNumber) ret;
  theNumber = value;
  print("OS Instances update done");
}

Snippet ID:	#1035030
Snippet name:	OS Instances Connector [backup]
Eternal ID of this version:	#1035030/1
Text MD5:	e401d37e10f5a35b068846141bd97b54
Author:	stefan
Category:	javax / a.i. / web
Type:	JavaX module (desktop)
Public (visible to everyone):	Yes
Archived (hidden from active list):	No
Created/modified:	2022-03-24 13:24:56
Source code size:	12300 bytes / 433 lines
Pitched / IR pitched:	No / No
Views / Downloads:	77 / 172
Referenced in:	[show references]

< > BotCompany Repo | #1035030 // OS Instances Connector [backup]

JavaX module (desktop)

Author comment

1	!7
2
3	static int updateInterval = 10;
4	static long timeout = 30000+10000; // idle time + latency
5	static int port = 6000;
6	static int maxLineLength = 1010241024; // change #1020500 too if you change this
7
8	import java.nio.*;
9	import java.nio.channels.*;
10
11	static ConnectedInstances cI;
12	static ReliableSingleThread rst = new(r update);
13	static int theNumber;
14	static Map<SocketChannel, State> dataMapper = synchroMap();
15	static ServerSocketChannel serverChannel;
16	static Selector selector;
17	static Map<S, Channel> channels = synchroMap();
18	static Lock channelsLock = lock();
19	static java.util.Timer mainTimer;
20	sbool debug;
21
22	sclass State {
23	S socketID = aGlobalID(), computerID;
24	int countSeen = -1;
25	new LineBuffer lineBuffer;
26	long lastMessage = sysNow();
27	Map<S, Channel> subbedChannels = synchroLinkedHashMap();
28	Lock lock = lock();
29
30	// TODO: clear eventually to avoid stalling attacks
31	L<byte[]> toSend = synchroLinkedList();
32	}
33
34	sclass Channel<A> {
35	S name;
36	Set<SocketChannel> sockets = synchroSet();
37	Map<SocketChannel, A> dataMapper = synchroMap();
38	Lock lock = lock();
39
40	void addSocket(SocketChannel socket) {
41	lock lock;
42	sockets.add(socket);
43	initialMessage(socket);
44	}
45
46	void removeSocket(SocketChannel socket) {
47	lock lock;
48	sockets.remove(socket);
49	}
50
51	void update() {}
52	void initialMessage(SocketChannel socket) {}
53	void handleLine(S line, SocketChannel socket) {}
54	}
55
56	sclass ComputerIDsChannel extends Channel<O> {
57	L<S> value;
58
59	void initialMessage(SocketChannel socket) {
60	lock lock;
61	sendTo(socket);
62	}
63
64	void update() {
65	if (set_trueIfChanged(this, value := asList(cI.set())))
66	for (SocketChannel s : sockets) sendTo(s);
67	}
68
69	void sendTo(SocketChannel socket) {
70	sendLineToSocket(socket, "ComputerIDs = " + struct(value));
71	}
72	}
73
74	sclass ComputerCountChannel extends Channel<O> {
75	int value;
76
77	void initialMessage(SocketChannel socket) {
78	lock lock;
79	sendTo(socket);
80	}
81
82	void update() {
83	if (set_trueIfChanged(this, value := cI.value()))
84	for (SocketChannel s : sockets) sendTo(s);
85	}
86
87	void sendTo(SocketChannel socket) {
88	sendLineToSocket(socket, "ComputerCount = " + value);
89	}
90	}
91
92	sclass PublicChatChannel extends Channel<O> {
93	L<S> msgs = synchroLinkedList();
94
95	void handleLine(S line, SocketChannel socket) {
96	postMessage(unquote(line));
97	}
98
99	void postMessage(S msg) {
100	msgs.add(msg);
101	}
102
103	void update {
104	S msg;
105	while ((msg = syncPopFirst(msgs)) != null) {
106	S quoted = quote(msg);
107	//print("Chat send: " + shorten(quoted, 20));
108	quoted = name + ":" + quoted;
109	for (SocketChannel s : sockets)
110	sendLineToSocket(s, quoted);
111	}
112	}
113	}
114
115	sclass PrivateChatChannel extends Channel<O> {
116	S prefix = "";
117	new MultiMap<S, SocketChannel> clients; // key: computer ID
118
119	void initialMessage(SocketChannel socket) {
120	clients.put(computerID(socket), socket);
121	}
122
123	void removeSocket(SocketChannel socket) {
124	super.removeSocket(socket);
125	clients.remove(computerID(socket), socket);
126	}
127
128	void handleLine(S line, SocketChannel socket) {
129	pcall {
130	int i = indexOf(line, ' ');
131	S computerID = takeFirst(line, i);
132	while licensed {
133	SocketChannel socket2 = clients.getFirst(computerID);
134	if (socket2 == null) break;
135	try {
136	S sender = computerID(socket);
137	if (l(line) >= 10000)
138	print("Forwarding line from " + sender + " to " + computerID + ": " + l(line));
139	sendLineToSocket(socket2, prefix + "privately from " + sender + ": " + substring(line, i+1));
140	break;
141	} catch e {
142	closeSocket(socket2);
143	clients.remove(computerID, socket2);
144	}
145	}
146	}
147	}
148	}
149
150	sclass SnippetUpdatesChannel extends Channel<O> {
151	ManualTailFile tail;
152
153	*() {
154	tail = manualTailFile_newOnly(programFile(#1019175, "changes.log"),
155	vfAppendToLineBuffer(unquotingLineBuffer(voidfunc(S line) { sendLine(line) })));
156	}
157
158	void update {
159	tail.check();
160	}
161
162	void sendLine(S line) {
163	print("SnippetUpdatesChannel send: " + line);
164	line = assertNoLineBreaks(line);
165	print("Sockets: " + l(sockets));
166	//S quoted = quote(line);
167	for (SocketChannel s : sockets)
168	sendLineToSocket(s, "snippetUpdates:" + line);
169	}
170	}
171
172	static State state(SocketChannel socket) {
173	ret dataMapper.get(socket);
174	}
175
176	sS computerID(SocketChannel socket) {
177	State state = state(socket);
178	ret state == null ? null : state.computerID;
179	}
180
181	p {
182	cI = new ConnectedInstances;
183	cI.setAfterglowInterval(1000);
184	cI.connected2.onChange(rst);
185	cI.start();
186
187	openServerSocket();
188
189	// restart socket every hour (and once now for testing - XX)
190	/doEvery(10.0, 6060.0, r {
191	pcall { cleanMeUp_socket(); }
192	openServerSocket();
193	});*/
194	}
195
196	svoid openServerSocket ctex {
197	print("Opening server socket");
198	selector = Selector.open();
199	serverChannel = ServerSocketChannel.open();
200	serverChannel.configureBlocking(false);
201
202	InetSocketAddress sockAddr = new InetSocketAddress(port);
203	serverChannel.socket().bind(sockAddr);
204	print("Server socket bound");
205	serverChannel.register(selector, SelectionKey.OP_ACCEPT);
206
207	thread "Server Accept Loop" { acceptLoop(); }
208
209	mainTimer = doEvery(updateInterval, r {
210	for (SocketChannel socket : keysList(dataMapper)) {
211	State state = state(socket);
212	if (state != null && sysTime() >= state.lastMessage+timeout) pcall {
213	closeSocket(socket);
214	}
215	}
216	for (Channel channel : values(channels)) pcall {
217	lock channel.lock;
218	channel.update();
219	}
220	});
221	}
222
223	svoid cleanMeUp_socket() ctex {
224	if (serverChannel != null)
225	serverChannel.close();
226	for (SocketChannel socket : keys(getAndClearMap(dataMapper))) pcall {
227	socket.close();
228	}
229	}
230
231	svoid acceptLoop ctex {
232	while licensed {
233	// wait for events
234	selector.select();
235
236	Iterator<SelectionKey> keys = selector.selectedKeys().iterator();
237	while (keys.hasNext()) {
238	SelectionKey key = keys.next();
239	keys.remove();
240	debug "have key";
241
242	if (!key.isValid())
243	continue with debug "key not valid";
244
245	pcall {
246	if (key.isAcceptable()) {
247	debug "Key is acceptable";
248	acceptOnSocket(key);
249	} else if (key.isReadable()) {
250	debug "Key is readable";
251	readFromSocket(key);
252	} else if (key.isWritable()) {
253	debug "Key is writable";
254	writeToSocket(key);
255	} else
256	debug "Unknown key type";
257	}
258	}
259	}
260	}
261
262	//accept a connection made to this channel's socket
263	static void acceptOnSocket(SelectionKey key) throws IOException {
264	ServerSocketChannel serverChannel = cast key.channel();
265	SocketChannel channel = serverChannel.accept();
266	channel.configureBlocking(false);
267	Socket actualSocket = channel.socket();
268	SocketAddress remoteAddr = actualSocket.getRemoteSocketAddress();
269	print("Connected to: " + remoteAddr);
270
271	// register channel with selector for further IO
272	dataMapper.put(channel, new State);
273	channel.register(selector, SelectionKey.OP_READ);
274	}
275
276	svoid onSocketClose(SocketChannel socket) {
277	Socket actualSocket = socket.socket();
278	SocketAddress remoteAddr = actualSocket.getRemoteSocketAddress();
279	print("Connection closed by client: " + remoteAddr);
280	State state = state(socket);
281	if (state != null && state.computerID != null)
282	cI.lostConnection(state.computerID);
283	dataMapper.remove(socket);
284	removeFromAllChannels(socket);
285	}
286
287	svoid closeSocket(SocketChannel socket) {
288	pcall { socket.close(); }
289	pcall { onSocketClose(socket); }
290	}
291
292	static void readFromSocket(SelectionKey key) throws IOException {
293	final SocketChannel socket = cast key.channel();
294	ByteBuffer buffer = ByteBuffer.allocate(1024);
295	int numRead = socket.read(buffer);
296	if (numRead == -1) {
297	closeSocket(socket);
298	key.cancel();
299	ret;
300	}
301
302	S s = new String(buffer.array(), 0, numRead);
303	//print("Got data from client: " + s);
304	final State state = state(socket);
305	if (state != null) {
306	state.lastMessage = sysNow();
307	state.lineBuffer.append(s, voidfunc(S line) {
308	//if (nempty(line)) print("Got line: " + line);
309	pcall { handleLine(line, socket, state); }
310	});
311	if (state.lineBuffer.size() > maxLineLength) {
312	print("Bad client (line too long");
313	closeSocket(socket);
314	}
315	}
316	}
317
318	static void writeToSocket(SelectionKey key) throws IOException {
319	final SocketChannel socket = cast key.channel();
320	socket.register(selector, SelectionKey.OP_READ);
321	State state = state(socket);
322	lock state.lock;
323	if (empty(state.toSend)) ret;
324	byte[] data = popFirst(state.toSend);
325	print("Sending postponed data (" + n2(l(data), "byte") + ") to " + state.computerID);
326	sendBytesToSocket(socket, data);
327	}
328
329	svoid sendBytesToSocket(SocketChannel socket, byte[] bytes) throws IOException {
330	int length = l(bytes);
331	State state = state(socket);
332	lock state.lock;
333	if (nempty(state.toSend)) {
334	print("Postponing send to " + state.computerID + " because data already queued");
335	ret with state.toSend.add(bytes);
336	}
337	ByteBuffer buf = ByteBuffer.wrap(bytes);
338	int n = socket.write(buf);
339	if (n < length) {
340	byte[] remaining = subByteArray(bytes, n);
341	state.toSend.add(remaining);
342	print("Postponing sending " + n2(l(remaining), "byte") + " to " + state.computerID);
343	socket.register(selector, SelectionKey.OP_READ \| SelectionKey.OP_WRITE);
344	}
345	}
346
347	svoid handleLine(S line, SocketChannel socket, State state) {
348	new Matches m;
349	if (empty(line)) ret with sendLineToSocket(socket, ""); // idle
350
351	if (jmatch("computerID=*", line, m)) {
352	if (state.computerID != null \|\| !possibleComputerID($1)) ret;
353	print("Got computer ID: " + $1);
354	state.computerID = $1;
355	cI.gotConnection(state.computerID);
356	ret;
357	}
358
359	if (jmatch("sub *", line, m)) {
360	Channel channel = getChannelForSocket(socket, $1);
361	if (channel == null)
362	sendLineToSocket(socket, "Channel " + $1 + " not found");
363	else
364	sendLineToSocket(socket, "Subbed " + $1);
365	ret;
366	}
367
368	L<S> tok = javaTok(line);
369	if (eq(get(tok, 3), ":")) {
370	Channel channel = state.subbedChannels.get(get(tok, 1));
371	if (channel != null)
372	channel.handleLine(joinSubList(tok, 5), socket);
373	}
374
375	// ignore unknown line
376	}
377
378	html { ret "Use port " + port; }
379
380	svoid sendLineToSocket(SocketChannel socket, S s) ctex {
381	byte[] bytes = toUtf8(s + "\n");
382	sendBytesToSocket(socket, bytes);
383	}
384
385	svoid removeFromAllChannels(SocketChannel socket) {
386	lock channelsLock;
387	for (S name, Channel c : cloneMap(channels)) { pcall {
388	c.removeSocket(socket);
389	if (empty(c.sockets)) {
390	print("Deleting channel: " + name);
391	channels.remove(name);
392	}
393	}}
394	}
395
396	static Channel getChannelForSocket(SocketChannel socket, S name) {
397	lock channelsLock;
398	Channel c = channels.get(name);
399	if (c == null) {
400	c = makeChannel(name);
401	c.name = name;
402	if (c == null) null;
403	print("New channel: " + name);
404	channels.put(name, c);
405	lock c.lock;
406	c.update();
407	}
408	c.addSocket(socket);
409	state(socket).subbedChannels.put(name, c);
410	ret c;
411	}
412
413	static Channel makeChannel(S name) {
414	if (eq(name, "computerIDs")) ret new ComputerIDsChannel;
415	if (eq(name, "computerCount")) ret new ComputerCountChannel;
416	if (eq(name, "chat")) ret new PublicChatChannel;
417	if (eq(name, "machineChat")) ret new PublicChatChannel;
418	if (eq(name, "privateChat")) ret new PrivateChatChannel;
419	if (eq(name, "privateMachineChat")) ret nu(PrivateChatChannel, prefix := name + ":");
420	if (eq(name, "snippetUpdates")) ret new SnippetUpdatesChannel;
421
422	ret null with print("Unknown channel: " + name);
423	}
424
425	svoid update {
426	print("OS Instances update");
427	if (empty(dataMapper)) ret;
428
429	int value = cI.value();
430	//if (value == theNumber) ret;
431	theNumber = value;
432	print("OS Instances update done");
433	}