# HG changeset patch
# User morrowa@betelgeuse.local
# Date 1245782670 25200
# Node ID de59ce19f42e27f844a78508468797a17b3078f6
# Parent  20cccc7c26eed98989eb4281cbf4f033c5a9d494
BROKEN COMMIT. Majority of code to handle closing has been added. Listeners do not close correctly.

diff -r 20cccc7c26ee -r de59ce19f42e 1_0_to_1_1_diffs.diff
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/1_0_to_1_1_diffs.diff	Tue Jun 23 11:44:30 2009 -0700
@@ -0,0 +1,425 @@
+diff -r 70590cc555aa -r 16454d63d4c2 BLIP/BLIPConnection.h
+--- a/BLIP/BLIPConnection.h	Thu Jun 19 10:22:19 2008 -0700
++++ b/BLIP/BLIPConnection.h	Mon Jun 23 14:02:31 2008 -0700
+@@ -20,6 +20,7 @@
+ @interface BLIPConnection : TCPConnection
+ {
+     BLIPDispatcher *_dispatcher;
++    BOOL _blipClosing;
+ }
+ 
+ /** The delegate object that will be called when the connection opens, closes or receives messages. */
+@@ -73,6 +74,13 @@
+ /** Called when a BLIPResponse (to one of your requests) is received from the peer.
+     This is called <i>after</i> the response object's onComplete target, if any, is invoked.*/
+ - (void) connection: (BLIPConnection*)connection receivedResponse: (BLIPResponse*)response;
++
++/** Called when the peer wants to close the connection. Return YES to allow, NO to prevent. */
++- (BOOL) connectionReceivedCloseRequest: (BLIPConnection*)connection;
++
++/** Called if the peer refuses a close request. 
++    The typical error is kBLIPError_Forbidden. */
++- (void) connection: (BLIPConnection*)connection closeRequestFailedWithError: (NSError*)error;
+ @end
+ 
+ 
+diff -r 70590cc555aa -r 16454d63d4c2 BLIP/BLIPConnection.m
+--- a/BLIP/BLIPConnection.m	Thu Jun 19 10:22:19 2008 -0700
++++ b/BLIP/BLIPConnection.m	Mon Jun 23 14:02:31 2008 -0700
+@@ -8,6 +8,7 @@
+ 
+ #import "BLIPConnection.h"
+ #import "BLIP_Internal.h"
++#import "TCP_Internal.h"
+ #import "BLIPReader.h"
+ #import "BLIPWriter.h"
+ #import "BLIPDispatcher.h"
+@@ -15,6 +16,7 @@
+ #import "Logging.h"
+ #import "Test.h"
+ #import "ExceptionUtils.h"
++#import "Target.h"
+ 
+ 
+ NSString* const BLIPErrorDomain = @"BLIP";
+@@ -33,10 +35,14 @@
+ }
+ 
+ 
++@interface BLIPConnection ()
++- (void) _handleCloseRequest: (BLIPRequest*)request;
++@end
+ 
+ 
+ @implementation BLIPConnection
+ 
++
+ - (void) dealloc
+ {
+     [_dispatcher release];
+@@ -48,6 +54,11 @@
+ - (id<BLIPConnectionDelegate>) delegate                     {return (id)_delegate;}
+ - (void) setDelegate: (id<BLIPConnectionDelegate>)delegate  {_delegate = delegate;}
+ 
++
++#pragma mark -
++#pragma mark RECEIVING:
++
++
+ - (BLIPDispatcher*) dispatcher
+ {
+     if( ! _dispatcher ) {
+@@ -58,11 +69,23 @@
+ }
+ 
+ 
++- (void) _dispatchMetaRequest: (BLIPRequest*)request
++{
++    NSString* profile = request.profile;
++    if( [profile isEqualToString: kBLIPProfile_Bye] )
++        [self _handleCloseRequest: request];
++    else
++        [request respondWithErrorCode: kBLIPError_NotFound message: @"Unknown meta profile"];
++}
++
++
+ - (void) _dispatchRequest: (BLIPRequest*)request
+ {
+     LogTo(BLIP,@"Received all of %@",request.descriptionWithProperties);
+     @try{
+-        if( ! [self.dispatcher dispatchMessage: request] )
++        if( request._flags & kBLIP_Meta )
++            [self _dispatchMetaRequest: request];
++        else if( ! [self.dispatcher dispatchMessage: request] )
+             [self tellDelegate: @selector(connection:receivedRequest:) withObject: request];
+         if( ! request.noReply && ! request.repliedTo ) {
+             LogTo(BLIP,@"Returning default empty response to %@",request);
+@@ -81,6 +104,10 @@
+ }
+ 
+ 
++#pragma mark -
++#pragma mark SENDING:
++
++
+ - (BLIPRequest*) request
+ {
+     return [[[BLIPRequest alloc] _initWithConnection: self body: nil properties: nil] autorelease];
+@@ -103,11 +130,61 @@
+ }
+ 
+ 
++#pragma mark -
++#pragma mark CLOSING:
++
++
++- (void) _beginClose
++{
++    // Override of TCPConnection method. Instead of closing the socket, send a 'bye' request:
++    if( ! _blipClosing ) {
++        LogTo(BLIPVerbose,@"Sending close request...");
++        BLIPRequest *r = [self request];
++        [r _setFlag: kBLIP_Meta value: YES];
++        r.profile = kBLIPProfile_Bye;
++        BLIPResponse *response = [r send];
++        response.onComplete = $target(self,_receivedCloseResponse:);
++    }
++    // Put the writer in close mode, to prevent client from sending any more requests:
++    [self.writer close];
++}
++
++- (void) _receivedCloseResponse: (BLIPResponse*)response
++{
++    NSError *error = response.error;
++    LogTo(BLIPVerbose,@"Received close response: error=%@",error);
++    if( error ) {
++        [self _unclose];
++        [self tellDelegate: @selector(connection:closeRequestFailedWithError:) withObject: error];
++    } else {
++        // Now finally close the socket:
++        [super _beginClose];
++    }
++}
++
++
++- (void) _handleCloseRequest: (BLIPRequest*)request
++{
++    LogTo(BLIPVerbose,@"Received a close request");
++    if( [_delegate respondsToSelector: @selector(connectionReceivedCloseRequest:)] )
++        if( ! [_delegate connectionReceivedCloseRequest: self] ) {
++            LogTo(BLIPVerbose,@"Responding with denial of close request");
++            [request respondWithErrorCode: kBLIPError_Forbidden message: @"Close request denied"];
++            return;
++        }
++    
++    LogTo(BLIPVerbose,@"Close request accepted");
++    _blipClosing = YES; // this prevents _beginClose from sending a close request back
++    [self close];
++}
++
++
+ @end
+ 
+ 
+ 
+ 
++#pragma mark -
+ @implementation BLIPListener
+ 
+ - (id) initWithPort: (UInt16)port
+diff -r 70590cc555aa -r 16454d63d4c2 BLIP/BLIPMessage.m
+--- a/BLIP/BLIPMessage.m	Thu Jun 19 10:22:19 2008 -0700
++++ b/BLIP/BLIPMessage.m	Mon Jun 23 14:02:31 2008 -0700
+@@ -74,6 +74,8 @@
+         [desc appendString: @", urgent"];
+     if( _flags & kBLIP_NoReply )
+         [desc appendString: @", noreply"];
++    if( _flags & kBLIP_Meta )
++        [desc appendString: @", META"];
+     [desc appendString: @"]"];
+     return desc;
+ }
+@@ -103,6 +105,8 @@
+         _flags &= ~flag;
+ }
+ 
++- (BLIPMessageFlags) _flags                 {return _flags;}
++
+ - (BOOL) compressed                         {return (_flags & kBLIP_Compressed) != 0;}
+ - (BOOL) urgent                             {return (_flags & kBLIP_Urgent) != 0;}
+ - (void) setCompressed: (BOOL)compressed    {[self _setFlag: kBLIP_Compressed value: compressed];}
+diff -r 70590cc555aa -r 16454d63d4c2 BLIP/BLIPReader.m
+--- a/BLIP/BLIPReader.m	Thu Jun 19 10:22:19 2008 -0700
++++ b/BLIP/BLIPReader.m	Mon Jun 23 14:02:31 2008 -0700
+@@ -93,7 +93,7 @@
+ 
+ - (BOOL) isBusy
+ {
+-    return _curBytesRead > 0;
++    return _curBytesRead > 0 || _pendingRequests.count > 0 || _pendingResponses.count > 0;
+ }
+ 
+ 
+diff -r 70590cc555aa -r 16454d63d4c2 BLIP/BLIPRequest.m
+--- a/BLIP/BLIPRequest.m	Thu Jun 19 10:22:19 2008 -0700
++++ b/BLIP/BLIPRequest.m	Mon Jun 23 14:02:31 2008 -0700
+@@ -199,6 +199,8 @@
+         setObj(&_mutableBody,nil);
+         
+         BLIPMutableProperties *errorProps = [self.properties mutableCopy];
++        if( ! errorProps )
++            errorProps = [[BLIPMutableProperties alloc] init];
+         NSDictionary *userInfo = error.userInfo;
+         for( NSString *key in userInfo ) {
+             id value = $castIf(NSString,[userInfo objectForKey: key]);
+@@ -227,8 +229,12 @@
+ {
+     Assert(_connection,@"%@ has no connection to send over",self);
+     Assert(!_sent,@"%@ was already sent",self);
++    BLIPWriter *writer = (BLIPWriter*)_connection.writer;
++    Assert(writer,@"%@'s connection has no writer (already closed?)",self);
+     [self _encode];
+-    return (self.sent = [(BLIPWriter*)_connection.writer sendMessage: self]);
++    BOOL sent = self.sent = [writer sendMessage: self];
++    Assert(sent);
++    return sent;
+ }
+ 
+ 
+diff -r 70590cc555aa -r 16454d63d4c2 BLIP/BLIPTest.m
+--- a/BLIP/BLIPTest.m	Thu Jun 19 10:22:19 2008 -0700
++++ b/BLIP/BLIPTest.m	Mon Jun 23 14:02:31 2008 -0700
+@@ -35,6 +35,7 @@
+ #define kClientUsesSSLCert          NO
+ #define kListenerRequiresSSL        NO
+ #define kListenerRequiresClientCert NO
++#define kListenerCloseAfter         50
+ 
+ 
+ static SecIdentityRef GetClientIdentity(void) {
+@@ -100,36 +101,38 @@
+ 
+ - (void) sendAMessage
+ {
+-    if(_pending.count<100) {
+-        Log(@"** Sending another %i messages...", kNBatchedMessages);
+-        for( int i=0; i<kNBatchedMessages; i++ ) {
+-            size_t size = random() % 32768;
+-            NSMutableData *body = [NSMutableData dataWithLength: size];
+-            UInt8 *bytes = body.mutableBytes;
+-            for( size_t i=0; i<size; i++ )
+-                bytes[i] = i % 256;
+-            
+-            BLIPRequest *q = [_conn requestWithBody: body
+-                                         properties: $dict({@"Content-Type", @"application/octet-stream"},
+-                                                           {@"User-Agent", @"BLIPConnectionTester"},
+-                                                           {@"Date", [[NSDate date] description]},
+-                                                           {@"Size",$sprintf(@"%u",size)})];
+-            Assert(q);
+-            if( kUseCompression && (random()%2==1) )
+-                q.compressed = YES;
+-            if( random()%16 > 12 )
+-                q.urgent = YES;
+-            BLIPResponse *response = [q send];
+-            Assert(response);
+-            Assert(q.number>0);
+-            Assert(response.number==q.number);
+-            [_pending setObject: $object(size) forKey: $object(q.number)];
+-            response.onComplete = $target(self,responseArrived:);
++    if( _conn.status==kTCP_Open || _conn.status==kTCP_Opening ) {
++        if(_pending.count<100) {
++            Log(@"** Sending another %i messages...", kNBatchedMessages);
++            for( int i=0; i<kNBatchedMessages; i++ ) {
++                size_t size = random() % 32768;
++                NSMutableData *body = [NSMutableData dataWithLength: size];
++                UInt8 *bytes = body.mutableBytes;
++                for( size_t i=0; i<size; i++ )
++                    bytes[i] = i % 256;
++                
++                BLIPRequest *q = [_conn requestWithBody: body
++                                             properties: $dict({@"Content-Type", @"application/octet-stream"},
++                                                               {@"User-Agent", @"BLIPConnectionTester"},
++                                                               {@"Date", [[NSDate date] description]},
++                                                               {@"Size",$sprintf(@"%u",size)})];
++                Assert(q);
++                if( kUseCompression && (random()%2==1) )
++                    q.compressed = YES;
++                if( random()%16 > 12 )
++                    q.urgent = YES;
++                BLIPResponse *response = [q send];
++                Assert(response);
++                Assert(q.number>0);
++                Assert(response.number==q.number);
++                [_pending setObject: $object(size) forKey: $object(q.number)];
++                response.onComplete = $target(self,responseArrived:);
++            }
++        } else {
++            Warn(@"There are %u pending messages; waiting for the listener to catch up...",_pending.count);
+         }
+-    } else {
+-        Warn(@"There are %u pending messages; waiting for the listener to catch up...",_pending.count);
++        [self performSelector: @selector(sendAMessage) withObject: nil afterDelay: kSendInterval];
+     }
+-    [self performSelector: @selector(sendAMessage) withObject: nil afterDelay: kSendInterval];
+ }
+ 
+ - (void) responseArrived: (BLIPResponse*)response
+@@ -191,6 +194,13 @@
+     Log(@"Now %u replies pending", _pending.count);
+ }
+ 
++- (BOOL) connectionReceivedCloseRequest: (BLIPConnection*)connection
++{
++    BOOL response = NO;
++    Log(@"***** %@ received a close request; returning %i",connection,response);
++    return response;
++}
++
+ 
+ @end
+ 
+@@ -217,6 +227,7 @@
+ @interface BLIPTestListener : NSObject <TCPListenerDelegate, BLIPConnectionDelegate>
+ {
+     BLIPListener *_listener;
++    int _nReceived;
+ }
+ 
+ @end
+@@ -277,6 +288,7 @@
+ - (void) connectionDidOpen: (TCPConnection*)connection
+ {
+     Log(@"** %@ didOpen [SSL=%@]",connection,connection.actualSecurityLevel);
++    _nReceived = 0;
+ }
+ - (BOOL) connection: (TCPConnection*)connection authorizeSSLPeer: (SecCertificateRef)peerCert
+ {
+@@ -312,6 +324,22 @@
+     AssertEq([[request valueOfProperty: @"Size"] intValue], size);
+ 
+     [request respondWithData: body contentType: request.contentType];
++    
++    if( ++ _nReceived == kListenerCloseAfter ) {
++        Log(@"********** Closing BLIPTestListener after %i requests",_nReceived);
++        [connection close];
++    }
++}
++
++- (BOOL) connectionReceivedCloseRequest: (BLIPConnection*)connection;
++{
++    Log(@"***** %@ received a close request",connection);
++    return YES;
++}
++
++- (void) connection: (BLIPConnection*)connection closeRequestFailedWithError: (NSError*)error
++{
++    Log(@"***** %@'s close request failed: %@",connection,error);
+ }
+ 
+ 
+diff -r 70590cc555aa -r 16454d63d4c2 BLIP/BLIPWriter.m
+--- a/BLIP/BLIPWriter.m	Thu Jun 19 10:22:19 2008 -0700
++++ b/BLIP/BLIPWriter.m	Mon Jun 23 14:02:31 2008 -0700
+@@ -79,10 +79,6 @@
+ 
+ - (BOOL) sendMessage: (BLIPMessage*)message
+ {
+-    if( _shouldClose ) {
+-        Warn(@"%@: Attempt to send a message after the connection has started closing",self);
+-        return NO;
+-    }
+     Assert(!message.sent,@"message has already been sent");
+     [self _queueMessage: message isNew: YES];
+     return YES;
+@@ -91,12 +87,14 @@
+ 
+ - (BOOL) sendRequest: (BLIPRequest*)q response: (BLIPResponse*)response
+ {
+-    if( !_shouldClose ) {
+-        [q _assignedNumber: ++_numRequestsSent];
+-        if( response ) {
+-            [response _assignedNumber: _numRequestsSent];
+-            [(BLIPReader*)self.reader _addPendingResponse: response];
+-        }
++    if( _shouldClose ) {
++        Warn(@"%@: Attempt to send a request after the connection has started closing: %@",self,q);
++        return NO;
++    }
++    [q _assignedNumber: ++_numRequestsSent];
++    if( response ) {
++        [response _assignedNumber: _numRequestsSent];
++        [(BLIPReader*)self.reader _addPendingResponse: response];
+     }
+     return [self sendMessage: q];
+ }
+diff -r 70590cc555aa -r 16454d63d4c2 BLIP/BLIP_Internal.h
+--- a/BLIP/BLIP_Internal.h	Thu Jun 19 10:22:19 2008 -0700
++++ b/BLIP/BLIP_Internal.h	Mon Jun 23 14:02:31 2008 -0700
+@@ -29,6 +29,7 @@
+     kBLIP_Urgent    = 0x0020,       // please send sooner/faster
+     kBLIP_NoReply   = 0x0040,       // no RPY needed
+     kBLIP_MoreComing= 0x0080,       // More frames coming (Applies only to individual frame)
++    kBLIP_Meta      = 0x0100,       // Special message type, handled internally (hello, bye, ...)
+ };
+ typedef UInt16 BLIPMessageFlags;
+ 
+@@ -41,7 +42,10 @@
+     UInt16           size;          // total size of frame, _including_ this header
+ } BLIPFrameHeader;
+ 
+-#define kBLIPFrameHeaderMagicNumber 0x9B34F205
++#define kBLIPFrameHeaderMagicNumber 0x9B34F206
++
++#define kBLIPProfile_Hi  @"Hi"      // Used for Profile header in meta greeting message
++#define kBLIPProfile_Bye @"Bye"     // Used for Profile header in meta close-request message
+ 
+ 
+ @interface BLIPConnection ()
+@@ -52,6 +56,7 @@
+ 
+ @interface BLIPMessage ()
+ @property BOOL sent, propertiesAvailable, complete;
++- (BLIPMessageFlags) _flags;
+ - (void) _setFlag: (BLIPMessageFlags)flag value: (BOOL)value;
+ - (void) _encode;
+ @end
diff -r 20cccc7c26ee -r de59ce19f42e Python/BLIP.py
--- a/Python/BLIP.py	Sun May 24 15:03:39 2009 -0700
+++ b/Python/BLIP.py	Tue Jun 23 11:44:30 2009 -0700
@@ -27,7 +27,7 @@
 
 # INTERNAL CONSTANTS -- NO TOUCHIES!
 
-kFrameMagicNumber   = 0x9B34F205
+kFrameMagicNumber   = 0x9B34F206
 kFrameHeaderFormat  = '!LLHH'
 kFrameHeaderSize    = 12
 
@@ -36,11 +36,15 @@
 kMsgFlag_Urgent     = 0x0020
 kMsgFlag_NoReply    = 0x0040
 kMsgFlag_MoreComing = 0x0080
+kMsgFlag_Meta       = 0x0100
 
 kMsgType_Request    = 0
 kMsgType_Response   = 1
 kMsgType_Error      = 2
 
+kMsgProfile_Hi      = "Hi"
+kMsgProfile_Bye     = "Bye"
+
 
 log = logging.getLogger('BLIP')
 log.propagate = True
@@ -104,7 +108,7 @@
             self.connect(address)
         self.address = address
         self.listener = listener
-        self.onRequest = None
+        self.onRequest = self.onCloseRequest = self.onCloseRefused = None
         self.pendingRequests = {}
         self.pendingResponses = {}
         self.outBox = []
@@ -112,12 +116,7 @@
         self.inNumRequests = self.outNumRequests = 0
         self.sending = False
         self._endOfFrame()
-    
-    def close(self):
-        if self.status > kClosed:
-            self.status = kClosing
-            log.info("Connection closing...")
-        asynchat.async_chat.close(self)
+        self._closeWhenPossible = False
     
     def handle_connect(self):
         log.info("Connection open!")
@@ -130,25 +129,19 @@
         self.status = kDisconnected
         self.close()
     
-    def handle_close(self):
-        log.info("Connection closed!")
-        self.pendingRequests = self.pendingResponses = None
-        self.outBox = None
-        if self.status == kClosing:
-            self.status = kClosed
-        else:
-            self.status = kDisconnected
-        asynchat.async_chat.handle_close(self)
-        
     
     ### SENDING:
     
     @property
-    def canSend(self):
+    def isOpen(self):
         return self.status==kOpening or self.status==kOpen
     
+    @property
+    def canSend(self):
+        return self.isOpen and not self._closeWhenPossible
+    
     def _sendMessage(self, msg):
-        if self.canSend:
+        if self.isOpen:
             self._outQueueMessage(msg,True)
             if not self.sending:
                 log.debug("Waking up the output stream")
@@ -208,6 +201,7 @@
         else:
             log.debug("Nothing more to send")
             self.sending = False
+            self._closeIfReady()
             return None
     
     ### RECEIVING:
@@ -255,6 +249,7 @@
                 self.inNumRequests += 1
         elif msgType==kMsgType_Response or msgType==kMsgType_Error:
             message = self.pendingResponses.get(requestNo)
+            message._updateFlags(flags)
         
         if message != None:
             message._beginFrame(flags)
@@ -284,10 +279,88 @@
         try:
             msg._finished()
             if not msg.isResponse:
-                self.onRequest(msg)
+                if msg._meta:
+                    self._dispatchMetaRequest(msg)
+                else:
+                    self.onRequest(msg)
         except Exception, x:
             log.error("Exception handling incoming message: %s", traceback.format_exc())
             #FIX: Send an error reply
+        # Check to see if we're done and ready to close:
+        self._closeIfReady()
+    
+    def _dispatchMetaRequest(self, request):
+        """Handles dispatching internal meta requests."""
+        if request['Profile'] == kMsgProfile_Bye:
+            shouldClose = True
+            if self.onCloseRequest:
+                shouldClose = self.onCloseRequest()
+            if not shouldClose:
+                log.debug("Sending resfusal to close...")
+                response = request.response
+                response.isError = True
+                response['Error-Domain'] = "BLIP"
+                response['Error-Code'] = 403
+                response.body = "Close request denied"
+                response.send()
+            else:
+                log.debug("Sending permission to close...")
+                response = request.response
+                response.send()
+        else:
+            response = request.response
+            response.isError = True
+            response['Error-Domain'] = "BLIP"
+            response['Error-Code'] = 404
+            response.body = "Unknown meta profile"
+            response.send()
+    
+    ### CLOSING:
+    
+    def close(self):
+        """Publicly callable close method. Sends close request to peer."""
+        if self.status != kOpen:
+            return False
+        log.info("Sending close request...")
+        req = OutgoingRequest(self, None, {'Profile': kMsgProfile_Bye})
+        req._meta = True
+        req.response.onComplete = self._handleCloseResponse
+        if not req.send():
+            log.error("Error sending close request.")
+            return False
+        else:
+            self.status = kClosing
+        return True
+    
+    def _handleCloseResponse(self, response):
+        """Called when we receive a response to a close request."""
+        log.info("Received close response.")
+        if response.isError:
+            # remote refused to close
+            if self.onCloseRefused:
+                self.onCloseRefused(response)
+            self.status = kOpen
+        else:
+            # now wait until everything has finished sending, then actually close
+            log.info("No refusal, actually closing...")
+            self._closeWhenPossible = True
+    
+    def _closeIfReady(self):
+        """Checks if all transmissions are complete and then closes the actual socket."""
+        if self._closeWhenPossible and len(self.outBox) == 0 and len(self.pendingRequests) == 0 and len(self.pendingResponses) == 0:
+            # self._closeWhenPossible = False
+            log.debug("_closeIfReady closing.")
+            asynchat.async_chat.close(self)
+    
+    def handle_close(self):
+        """Called when the socket actually closes."""
+        log.info("Connection closed!")
+        self.pendingRequests = self.pendingResponses = None
+        self.outBox = None
+        if self.status == kClosing:
+            self.status = kClosed
+        else:
+            self.status = kDisconnected
 
 
 ### MESSAGE CLASSES:
@@ -305,13 +378,17 @@
     @property
     def flags(self):
         if self.isResponse:
-            flags = kMsgType_Response
+            if self.isError:
+                flags = kMsgType_Error
+            else:
+                flags = kMsgType_Response
         else:
             flags = kMsgType_Request
         if self.urgent:     flags |= kMsgFlag_Urgent
         if self.compressed: flags |= kMsgFlag_Compressed
         if self.noReply:    flags |= kMsgFlag_NoReply
         if self._moreComing:flags |= kMsgFlag_MoreComing
+        if self._meta:      flags |= kMsgFlag_Meta
         return flags
     
     def __str__(self):
@@ -322,6 +399,7 @@
         if self.compressed: s += " CMP"
         if self.noReply:    s += " NOR"
         if self._moreComing:s += " MOR"
+        if self._meta:      s += " MET"
         if self.body:       s += " %i bytes" %len(self.body)
         return s+"]"
     
@@ -352,11 +430,16 @@
     def __init__(self, connection, requestNo, flags):
         super(IncomingMessage,self).__init__(connection)
         self.requestNo  = requestNo
+        self._updateFlags(flags)
+        self.frames     = []
+    
+    def _updateFlags(self, flags):
         self.urgent     = (flags & kMsgFlag_Urgent) != 0
         self.compressed = (flags & kMsgFlag_Compressed) != 0
         self.noReply    = (flags & kMsgFlag_NoReply) != 0
         self._moreComing= (flags & kMsgFlag_MoreComing) != 0
-        self.frames     = []
+        self._meta      = (flags & kMsgFlag_Meta) != 0
+        self.isError    = (flags & kMsgType_Error) != 0
     
     def _beginFrame(self, flags):
         """Received a frame header."""
@@ -377,16 +460,18 @@
         if propSize>len(encoded): raise MessageException, "properties too long to fit"
         if propSize>2 and encoded[propSize-1] != '\000': raise MessageException, "properties are not nul-terminated"
         
-        proplist = encoded[2:propSize-1].split('\000')
+        if propSize > 2:
+            proplist = encoded[2:propSize-1].split('\000')
+        
+            if len(proplist) & 1: raise MessageException, "odd number of property strings"
+            for i in xrange(0,len(proplist),2):
+                def expand(str):
+                    if len(str)==1:
+                        str = IncomingMessage.__expandDict.get(str,str)
+                    return str
+                self.properties[ expand(proplist[i])] = expand(proplist[i+1])
+        
         encoded = encoded[propSize:]
-        if len(proplist) & 1: raise MessageException, "odd number of property strings"
-        for i in xrange(0,len(proplist),2):
-            def expand(str):
-                if len(str)==1:
-                    str = IncomingMessage.__expandDict.get(str,str)
-                return str
-            self.properties[ expand(proplist[i])] = expand(proplist[i+1])
-        
         # Decode the body:
         if self.compressed and len(encoded)>0:
             try:
@@ -411,7 +496,7 @@
     
     def __init__(self, connection, body=None, properties=None):
         Message.__init__(self,connection,body,properties)
-        self.urgent = self.compressed = self.noReply = False
+        self.urgent = self.compressed = self.noReply = self._meta = self.isError = False
         self._moreComing = True
     
     def __setitem__(self, key,val):
@@ -435,7 +520,7 @@
         propertiesSize = out.tell()
         assert propertiesSize<65536     #FIX: Return an error instead
         
-        body = self.body
+        body = self.body or ""
         if self.compressed:
             z = zlib.compressobj(6,zlib.DEFLATED,31)   # window size of 31 needed for gzip format
             out.write(z.compress(body))
diff -r 20cccc7c26ee -r de59ce19f42e Python/BLIPConnectionTest.py
--- a/Python/BLIPConnectionTest.py	Sun May 24 15:03:39 2009 -0700
+++ b/Python/BLIPConnectionTest.py	Tue Jun 23 11:44:30 2009 -0700
@@ -72,7 +72,8 @@
     
     def tearDown(self):
         self.connection.close()
+        asyncore.loop() # got to give it time to negotiate close; this call should exit eventually
 
 if __name__ == '__main__':
-    logging.basicConfig(level=logging.INFO)
+    logging.basicConfig(level=logging.DEBUG)
     unittest.main()
diff -r 20cccc7c26ee -r de59ce19f42e Python/BLIPListenerTest.py
--- a/Python/BLIPListenerTest.py	Sun May 24 15:03:39 2009 -0700
+++ b/Python/BLIPListenerTest.py	Tue Jun 23 11:44:30 2009 -0700
@@ -42,5 +42,5 @@
             logging.info("KeyboardInterrupt")
 
 if __name__ == '__main__':
-    logging.basicConfig(level=logging.INFO)
+    logging.basicConfig(level=logging.DEBUG)
     unittest.main()
diff -r 20cccc7c26ee -r de59ce19f42e Python/CloseTestPing.py
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/Python/CloseTestPing.py	Tue Jun 23 11:44:30 2009 -0700
@@ -0,0 +1,36 @@
+# CloseTestPing.py
+# Tests the closing negotiation facilities of the BLIP 1.1 protocol
+
+from BLIP import Connection, OutgoingRequest
+
+import unittest
+import asyncore
+import logging
+
+class CloseTestPing(unittest.TestCase):
+    
+    def handleCloseRefusal(self, resp):
+        logging.info("Close request was refused!")
+    
+    def setUp(self):
+        self.connection = Connection( ('localhost', 1337) )
+        self.connection.onCloseRefused = self.handleCloseRefusal
+    
+    def handleResponse(self, resp):
+        logging.info("Got response...")
+    
+    def testClose(self):
+        req = OutgoingRequest(self.connection, "Ping")
+        req.response.onComplete = self.handleResponse
+        req.send()
+        
+        asyncore.loop(timeout=1, count=5)
+        
+        self.connection.close()
+        
+        asyncore.loop()
+
+
+if __name__ == '__main__':
+    logging.basicConfig(level=logging.DEBUG)
+    unittest.main()
diff -r 20cccc7c26ee -r de59ce19f42e Python/CloseTestPong.py
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/Python/CloseTestPong.py	Tue Jun 23 11:44:30 2009 -0700
@@ -0,0 +1,38 @@
+# CloseTestPong.py
+# Tests the closing negotiation facilities of the BLIP 1.1 protocol
+
+from BLIP import Listener
+
+import logging
+import asyncore
+import unittest
+
+class CloseTestPong(unittest.TestCase):
+    
+    def shouldClose(self):
+        logging.info("Allowed to close.")
+        return True
+    
+    def handleConnection(self, conn):
+        logging.info("Accepted connection.")
+        conn.onCloseRequest = self.shouldClose
+    
+    def handleRequest(self, req):
+        resp = req.response
+        resp.body = "Pong"
+        resp.send()
+    
+    def testClose(self):
+        listen = Listener(1337)
+        listen.onConnected = self.handleConnection
+        listen.onRequest = self.handleRequest
+        
+        try:
+            asyncore.loop()
+        except KeyboardInterrupt:
+            pass
+
+
+if __name__ == '__main__':
+    logging.basicConfig(level=logging.DEBUG)
+    unittest.main()
diff -r 20cccc7c26ee -r de59ce19f42e Python/asynchatPing.py
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/Python/asynchatPing.py	Tue Jun 23 11:44:30 2009 -0700
@@ -0,0 +1,60 @@
+# asynchatPing
+# Uses asynchat
+# Not related to BLIP - just to aid in my understanding of what's going on
+# Sends "Ping", waits for "Pong"
+
+import socket
+import asyncore
+import asynchat
+
+kNumPings = 10
+
+class asynchatPing(asynchat.async_chat):
+    def __init__(self, address):
+        asynchat.async_chat.__init__(self)
+        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.connect(address)
+        self.set_terminator("Pong")
+        self.pingsSent = self.pongsGot = 0
+        self.donePing = self.donePong = False
+    
+    def handle_connect(self):
+        print "Connected"
+    
+    def handle_close(self):
+        print "Closed"
+        asynchat.async_chat.handle_close(self)
+    
+    def collect_incoming_data(self, data):
+        """discard data"""
+        pass
+    
+    def found_terminator(self):
+        """when we get a Pong"""
+        print "Received 'Pong'"
+        self.pongsGot += 1
+        if self.pongsGot == kNumPings:
+            print "Done ponging"
+            self.donePong = True
+            self.close_when_done()
+    
+    def ping(self):
+        if not self.donePing:
+            self.push("Ping")
+            print "Sent 'Ping'"
+            self.pingsSent += 1
+            if self.pingsSent == kNumPings:
+                print "Done pinging"
+                self.donePing = True
+    
+    def run(self):
+        timeout = 0
+        while not self.donePing:
+            self.ping()
+            asyncore.loop(timeout=timeout, count=1)
+        asyncore.loop()
+        print "Done!"
+
+if __name__ == '__main__':
+    ping = asynchatPing( ('localhost', 1337) )
+    ping.run()
diff -r 20cccc7c26ee -r de59ce19f42e Python/asynchatPong.py
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/Python/asynchatPong.py	Tue Jun 23 11:44:30 2009 -0700
@@ -0,0 +1,59 @@
+# asynchatPong
+# Listener using asynchat
+# Not related to BLIP - just to aid in my understanding of what's going on
+# Sends "Pong" when it gets "Ping"
+
+import sys
+import traceback
+import socket
+import asyncore
+import asynchat
+
+class asynchatPongListener(asyncore.dispatcher):
+    def __init__(self, port):
+        asyncore.dispatcher.__init__(self)
+        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.bind( ('', port) )
+        self.listen(2)
+        self.shouldAccept = True
+    
+    def handle_accept(self):
+        if self.shouldAccept:
+            sock, addr = self.accept()
+            self.conn = asynchatPong(sock, self)
+            self.shouldAccept = False
+    
+    def handle_error(self):
+        (typ,val,trace) = sys.exc_info()
+        print "Listener caught: %s %s\n%s" % (typ,val,traceback.format_exc())
+        self.close()
+    
+    def handle_close(self):
+        print "Listener got close"
+        asyncore.dispatcher.handle_close(self)
+
+class asynchatPong(asynchat.async_chat):
+    def __init__(self, socket, listener):
+        asynchat.async_chat.__init__(self, socket)
+        self._listener = listener
+        self.set_terminator("Ping")
+    
+    def collect_incoming_data(self, data):
+        """called when arbitrary amount of data arrives. we just eat it"""
+        pass
+    
+    def found_terminator(self):
+        """called when the terminator we set is found"""
+        print "Found 'Ping'"
+        self.push("Pong")
+        print "Sent 'Pong'"
+    
+    def handle_close(self):
+        print "Closed; closing listener"
+        self._listener.close()
+        asynchat.async_chat.handle_close(self)
+    
+
+if __name__ == '__main__':
+    pong = asynchatPongListener(1337)
+    asyncore.loop()