Prepare the HTTP library for inclusion in the standalone VM (step 2)

samples/chat/chat_server_lib.dart

 // Copyright (c) 2012, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 #library("chat_server");
 #import("dart:io");
 #import("dart:json");
 #import("dart:isolate");
 #import("http.dart");
 
@@ skipping 39 matching lines @@
       _serverPort = port;
       _start(hostAddress, tcpPort, listenBacklog);
     });
     // We can only guess this is the right URL. At least it gives a
     // hint to the user.
     print('Server starting http://${hostAddress}:${tcpPort}/');
   }
 
     void _start(String hostAddress, int tcpPort, int listenBacklog) {
     // Handle status messages from the server.
-    _statusPort.receive(
-        void _(var message, SendPort replyTo) {
-          String status = message.message;
-          print("Received status: $status");
-        });
+    _statusPort.receive((var message, SendPort replyTo) {
+      String status = message.message;
+      print("Received status: $status");
+    });
 
     // Send server start message to the server.
     var command = new ChatServerCommand.start(hostAddress,
                                               tcpPort,
                                               backlog: listenBacklog);
     _serverPort.send(command, _statusPort.toSendPort());
   }
 
   void shutdown() {
     // Send server stop message to the server.
@@ skipping 22 matching lines @@
   String _handle;
   String _sessionId;
   Date _lastActive;
 }
 
 
 class Message {
   static final int JOIN = 0;
   static final int MESSAGE = 1;
   static final int LEAVE = 2;
-  static final int TIMEOUT = 2;
+  static final int TIMEOUT = 3;
   static final List<String> _typeName =
       const [ "join", "message", "leave", "timeout"];
 
   Message.join(this._from)
       : _received = new Date.now(), _type = JOIN;
   Message(this._from, this._message)
       : _received = new Date.now(), _type = MESSAGE;
   Message.leave(this._from)
       : _received = new Date.now(), _type = LEAVE;
   Message.timeout(this._from)
@@ skipping 49 matching lines @@
     _activeUsers.remove(sessionId);
   }
 
   bool _addMessage(Message message) {
     message.messageNumber = _nextMessageNumber++;
     _messages.add(message);
 
     // Send the new message to all polling clients.
     List messages = new List();
     messages.add(message.toMap());
-    _callbacks.forEach(
-        void _(String sessionId, Function callback) {
-          callback(messages);
-        });
+    _callbacks.forEach((String sessionId, Function callback) {
+      callback(messages);
+    });
     _callbacks = new Map();
   }
 
   bool _userMessage(Map requestData) {
     String sessionId = requestData["sessionId"];
     User user = _userLookup(sessionId);
     if (user == null) return false;
     String handle = user.handle;
     String messageText = requestData["message"];
     if (messageText == null) return false;
@@ skipping 24 matching lines @@
   }
 
   void registerChangeCallback(String sessionId, var callback) {
     _callbacks[sessionId] = callback;
   }
 
   void _handleTimer(Timer timer) {
     Set inactiveSessions = new Set();
     // Collect all sessions which have not been active for some time.
     Date now = new Date.now();
-    _activeUsers.forEach(
-        void _(String sessionId, User user) {
-          if (user.idleTime(now).inMilliseconds > DEFAULT_IDLE_TIMEOUT) {
-            inactiveSessions.add(sessionId);
-          }
-        });
+    _activeUsers.forEach((String sessionId, User user) {
+      if (user.idleTime(now).inMilliseconds > DEFAULT_IDLE_TIMEOUT) {
+        inactiveSessions.add(sessionId);
+      }
+    });
     // Terminate the inactive sessions.
-    inactiveSessions.forEach(
-        void _(String sessionId) {
-          Function callback = _callbacks.remove(sessionId);
-          if (callback != null) callback(null);
-          User user = _activeUsers.remove(sessionId);
-          Message message = new Message.timeout(user);
-          _addMessage(message);
-        });
-
+    inactiveSessions.forEach((String sessionId) {
+      Function callback = _callbacks.remove(sessionId);
+      if (callback != null) callback(null);
+      User user = _activeUsers.remove(sessionId);
+      Message message = new Message.timeout(user);
+      _addMessage(message);
+    });
   }
 
   Map<String, User> _activeUsers;
   List<Message> _messages;
   int _nextMessageNumber;
   Map<String, Function> _callbacks;
 }
 
 
 class ChatServerCommand {
@@ skipping 125 matching lines @@
         if (extension == ".js") { mimeType = "application/javascript"; }
         if (extension == ".ico") { mimeType = "image/vnd.microsoft.icon"; }
         if (extension == ".png") { mimeType = "image/png"; }
       }
       response.setHeader("Content-Type", mimeType);
       // Get the length of the file for setting the Content-Length header.
       RandomAccessFile openedFile = file.openSync();
       response.contentLength = openedFile.lengthSync();
       openedFile.close();
       // Pipe the file content into the response.
-      file.openInputStream().pipe(response.outputStream);
+      file.openInputStreamSync().pipe(response.outputStream);
     } else {
       print("File not found: $fileName");
       _notFoundHandler(request, response);
     }
   }
 
   // Serve the not found page.
   void _notFoundHandler(HTTPRequest request, HTTPResponse response) {
     if (_notFoundPage == null) {
       _notFoundPage = notFoundPageHtml.charCodes();
     }
     response.statusCode = HTTPStatus.NOT_FOUND;
     response.setHeader("Content-Type", "text/html; charset=UTF-8");
     response.contentLength = _notFoundPage.length;
     response.outputStream.write(_notFoundPage);
     response.outputStream.close();
   }
 
   // Unexpected protocol data.
   void _protocolError(HTTPRequest request, HTTPResponse response) {
     response.statusCode = HTTPStatus.INTERNAL_SERVER_ERROR;
     response.contentLength = 0;
     response.outputStream.close();
   }
 
   // Join request:
   // { "request": "join",
   //   "handle": <handle> }
   void _joinHandler(HTTPRequest request, HTTPResponse response) {
-    void dataEndHandler(String data) {
+    StringBuffer body = new StringBuffer();

[Anders Johnsen, 2012/02/28 12:23:40]

It could be nice to have a StringBuffer.fromStream(...). The problem is however,
that they reside in two different libs. Could we have an
InputStream.toStringBuffer?

[Mads Ager (google), 2012/02/28 12:27:40]

I agree that we should get back to a situation where you can just get a callback
when all the data has been received and optionally decoded.

[Søren Gjesse, 2012/02/28 13:03:42]

As discussed offline I will add a simple way to get all data in one callback.

+    StringInputStream input = new StringInputStream(request.inputStream);
+    input.dataHandler = () => body.add(input.read());
+    input.closeHandler = () {
+      String data = body.toString();
       if (data != null) {
         var requestData = JSON.parse(data);
         if (requestData["request"] == "join") {
           String handle = requestData["handle"];
           if (handle != null) {
             // New user joining.
             User user = _topic._userJoined(handle);
 
             // Send response.
             Map responseData = new Map();
             responseData["response"] = "join";
             responseData["sessionId"] = user.sessionId;
             _sendJSONResponse(response, responseData);
           } else {
             _protocolError(request, response);
           }
         } else {
           _protocolError(request, response);
         }
       } else {
         _protocolError(request, response);
       }
-    }
-
-    // Register callback for full request data.
-    request.dataEnd = dataEndHandler;
+    };
   }
 
   // Leave request:
   // { "request": "leave",
   //   "sessionId": <sessionId> }
   void _leaveHandler(HTTPRequest request, HTTPResponse response) {
-    void dataEndHandler(String data) {
+    StringBuffer body = new StringBuffer();
+    StringInputStream input = new StringInputStream(request.inputStream);
+    input.dataHandler = () => body.add(input.read());
+    input.closeHandler = () {
+      String data = body.toString();
       var requestData = JSON.parse(data);
       if (requestData["request"] == "leave") {
         String sessionId = requestData["sessionId"];
         if (sessionId != null) {
           // User leaving.
           _topic._userLeft(sessionId);
 
           // Send response.
           Map responseData = new Map();
           responseData["response"] = "leave";
           _sendJSONResponse(response, responseData);
         } else {
           _protocolError(request, response);
         }
       } else {
         _protocolError(request, response);
       }
-    }
-
-    request.dataEnd = dataEndHandler;
+    };
   }
 
   // Message request:
   // { "request": "message",
   //   "sessionId": <sessionId>,
   //   "message": <message> }
   void _messageHandler(HTTPRequest request, HTTPResponse response) {
-    void dataEndHandler(String data) {
+    StringBuffer body = new StringBuffer();
+    StringInputStream input = new StringInputStream(request.inputStream);
+    input.dataHandler = () => body.add(input.read());
+    input.closeHandler = () {
+      String data = body.toString();
       _messageCount++;
       _messageRate.record(1);
       var requestData = JSON.parse(data);
       if (requestData["request"] == "message") {
         String sessionId = requestData["sessionId"];
         if (sessionId != null) {
           // New message from user.
           bool success = _topic._userMessage(requestData);
 
           // Send response.
           if (success) {
             Map responseData = new Map();
             responseData["response"] = "message";
             _sendJSONResponse(response, responseData);
           } else {
             _protocolError(request, response);
           }
         } else {
           _protocolError(request, response);
         }
       } else {
         _protocolError(request, response);
       }
-    }
-
-    request.dataEnd = dataEndHandler;
+    };
   }
 
   // Receive request:
   // { "request": "receive",
   //   "sessionId": <sessionId>,
   //   "nextMessage": <nextMessage>,
   //   "maxMessages": <maxMesssages> }
   void _receiveHandler(HTTPRequest request, HTTPResponse response) {
-    void dataEndHandler(String data) {
+    StringBuffer body = new StringBuffer();
+    StringInputStream input = new StringInputStream(request.inputStream);
+    input.dataHandler = () => body.add(input.read());
+    input.closeHandler = () {
+      String data = body.toString();
       var requestData = JSON.parse(data);
       if (requestData["request"] == "receive") {
         String sessionId = requestData["sessionId"];
         int nextMessage = requestData["nextMessage"];
         int maxMessages = requestData["maxMessages"];
         if (sessionId != null && nextMessage != null) {
 
           void sendResponse(messages) {
             // Send response.
             Map responseData = new Map();
@@ skipping 16 matching lines @@
           } else {
             sendResponse(messages);
           }
 
         } else {
           _protocolError(request, response);
         }
       } else {
         _protocolError(request, response);
       }
-    }
-
-    request.dataEnd = dataEndHandler;
+    };
   }
 
   void addHandler(String path,
                   void handler(HTTPRequest request, HTTPResponse response)) {
     _requestHandlers[path] = handler;
   }
 
   void main() {
     _logRequests = false;
     _topic = new Topic();
     _serverStart = new Date.now();
     _messageCount = 0;
     _messageRate = new Rate();
 
     // Start a timer for cleanup events.
     _cleanupTimer =
         new Timer.repeating((timer) => _topic._handleTimer(timer), 10000);
 
     // Start timer for periodic logging.
     void _handleLogging(Timer timer) {
       if (_logging) {
         print((_messageRate.rate).toString() +
                        " messages/s (total " +
                        _messageCount +
                        " messages)");
       }
     }
 
-    this.port.receive(
-        void _(var message, SendPort replyTo) {
-          if (message.isStart) {
-            _host = message.host;
-            _port = message.port;
-            _logging = message.logging;
-            replyTo.send(new ChatServerStatus.starting(), null);
-            _server = new HTTPServer();
-            try {
-              _server.listen(_host, _port, backlog: message.backlog);
-              _server.requestHandler = (HTTPRequest req, HTTPResponse rsp) =>
-                  _requestReceivedHandler(req, rsp);
-              replyTo.send(new ChatServerStatus.started(_server.port), null);
-              _loggingTimer = new Timer.repeating(_handleLogging, 1000);
-            } catch (var e) {
-              replyTo.send(new ChatServerStatus.error(e.toString()), null);
-            }
-          } else if (message.isStop) {
-            replyTo.send(new ChatServerStatus.stopping(), null);
-            stop();
-            replyTo.send(new ChatServerStatus.stopped(), null);
-          }
-        });
+    this.port.receive((var message, SendPort replyTo) {
+      if (message.isStart) {
+        _host = message.host;
+        _port = message.port;
+        _logging = message.logging;
+        replyTo.send(new ChatServerStatus.starting(), null);
+        _server = new HTTPServer();
+        try {
+          _server.listen(_host, _port, backlog: message.backlog);
+          _server.requestHandler = (HTTPRequest req, HTTPResponse rsp) =>
+              _requestReceivedHandler(req, rsp);
+          replyTo.send(new ChatServerStatus.started(_server.port), null);
+          _loggingTimer = new Timer.repeating(_handleLogging, 1000);
+        } catch (var e) {
+          replyTo.send(new ChatServerStatus.error(e.toString()), null);
+        }
+      } else if (message.isStop) {
+        replyTo.send(new ChatServerStatus.stopping(), null);
+        stop();
+        replyTo.send(new ChatServerStatus.stopped(), null);
+      }
+    });
   }
 
   stop() {
     _server.close();
     _cleanupTimer.cancel();
     this.port.close();
   }
 
   void _requestReceivedHandler(HTTPRequest request, HTTPResponse response) {
     if (_logRequests) {
@@ skipping 96 matching lines @@
     }
   }
 
   int _timeRange;
   List<int> _buckets;
   int _currentBucket;
   int _currentBucketTime;
   num _bucketTimeRange;
   int _sum;
 }