Berichten met label _changes

Last time around, we were able to store the OSGi log messages into a CouchDB. Now I want to tap into the _changes API that CouchDB offers. The options are simple, connect to the ‘_changes’ document in a CouchDB database and you will be presented with a log of all changes that have occurred on the database.

http://localhost:5984/db/_changes

{"results":[
{"seq":3,"id":"_design/app","changes":[{"rev":"2-0dc1002d841221e8d8d7f325b87ffd67"}]},
{"seq":196,"id":"994cf6817a74882cecda6425e8001a96","changes":[{"rev":"193-083ec5c1e1359789eb2cf09c46097fee"}]}
,
"last_seq":196}

For each document in the database only the last change is presented. This could result in quite a lot of information. It is therefore also possible to fetch changes while applying a server-side filter. For this to work, you first need to define a filter in a design document. Mine looks like this:

function(doc, req) {
    if (doc.type == "EVENT") { return true; }
    return false;
}

and is stored (using couchapp) in a design document called ‘app’. The name of the filter is ‘events’. The filters’ meaning should obvious: pass through all documents for which the function returns ‘true’. The proper URL for accessing the changes will now be:

http://localhost:5984/db/_changes?filter=app/events

{"results":[
{"seq":196,"id":"994cf6817a74882cecda6425e8001a96","changes":[{"rev":"193-083ec5c1e1359789eb2cf09c46097fee"}]}
],
"last_seq":196}

The “last_seq” part can be used for bookkeeping: If you plan to make  multiple calls to the _changes feed (say every hour) then it is nice to be able to skip the ones you already saw during a previous call. So, the second time you call the _changes feed, you can skip all previous changes by employing the following URL:

http://localhost:5984/db/_changes?filter=app/events&since=196

{"results":[

],
"last_seq":196}

Obviously, no update has occurred since we accessed the _changes feed for the first time. If, however we now add a new document to the database (with doc.type == “EVENT”) and call the _changes feed again:

http://localhost:5984/db/_changes?filter=app/events&since=196

{"results":[
{"seq":197,"id":"f0a4a559fac65ab3cca87baf1c014fa7","changes":[{"rev":"1-c8dfffec4e08ca8db2b95f30613e213d"}]}
],
"last_seq":197}

Also, a call to “http://localhost:5984/db” will result in a JSON object, where “update_seq” provides the current update sequence. This comes in handy if you’re really not interested in previous changes:

http://localhost:5984/db

{"db_name":"db","doc_count":4,"doc_del_count":0,"update_seq":196,
"purge_seq":0,"compact_running":false,"disk_size":1871961,
"instance_start_time":"1282049630257719","disk_format_version":5,
"committed_update_seq":196}

This is all great functionality if your time granularity is in the order of hours or days. But hardly optimal if you want to be notified immediately of changes to the database. Fortunately, by supplying “feed=continuous” to the URL, CouchDB will keep your socket connection open and supply the changes as soon as they become available. Each change is represented by a well-formatted JSON line. This breaks from the previous examples, where the complete response consisted of well-formatted JSON, so beware!

http://localhost:5984/db/_changes?filter=app/events&since=196&feed=continuous

{"seq":197,"id":"f0a4a559fac65ab3cca87baf1c014fa7","changes":[{"rev":"1-c8dfffec4e08ca8db2b95f30613e213d"}]}
...
...
{"last_seq":197}

Eventually, if no changes have been sent for 60 secs, the socket connection closes. Before it does so, it spits out the last_seq, again for bookkeeping purposes.

I was wondering if it would be possible to create a simple iPhone/iPad application that is able to report specific events on a MapView using the _changes feed from CouchDB. All one then has to do as iPhone/ iPad application is connect to the continuous feed, parse the JSON data and display the events on your MKMapView. Server-side, all one has to do is insert/ update documents in the database. How easy!

So I started off by connecting to the _changes feed using an asynchronous NSURLConnection, expecting that its delegate would receive connection:didReceiveData: calls on each change. Unfortunately, this did not work: The NSURLConnection employs buffering internally that cannot be circumvented. Thus only after about 5 or 6 updates the delegate received some data. This is definitely not good enough. Some searching pointed me to the CFSocket/NSStream class. This class fortunately employs some more basic (read: unbuffered) socket handling, allowing me to receive data from the socket as soon as it becomes available!

    //open the socket
    CFReadStreamRef readStream;
    CFWriteStreamRef writeStream;
    CFStreamCreatePairWithSocketToHost(NULL, (CFStringRef)[theURL host], [[theURL port] intValue], &readStream, &writeStream);

    inputStream = (NSInputStream *)readStream;
    outputStream = (NSOutputStream *)writeStream;

    [inputStream setDelegate: self];
    [outputStream setDelegate :self];
    [inputStream scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
    [outputStream scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
    [inputStream open];
    [outputStream open];    

The containing class is set as delegate for both the NSInputStream and NSOutputStream. This causes the following method being called, based on events that occur on the streams:

/* stream eventing */
- (void)stream:(NSStream *)stream handleEvent:(NSStreamEvent)eventCode {

    switch(eventCode) {
        case NSStreamEventOpenCompleted:
            [self connection: nil didReceiveResponse: nil];
			break;

		case NSStreamEventErrorOccurred:
            [self connection: nil didFailWithError: [stream streamError]];
			break;

		case NSStreamEventEndEncountered:
            [stream close];
            [stream removeFromRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
            [stream release];
            inputStream = nil;
            outputStream = nil;
            [self connectionDidFinishLoading: nil];
			break;            

        case NSStreamEventHasSpaceAvailable:
        {
            if (stream == outputStream) {
                [self connection: nil willSendRequest: [NSURLRequest requestWithURL: theURL] redirectResponse: nil];

                NSString *str = [NSString stringWithFormat: @"GET %@?%@ HTTP/1.0\r\n\r\n", [theURL path], [theURL query]];
                const uint8_t *rawstring = (const uint8_t *)[str UTF8String];
                [outputStream write:rawstring maxLength: [str length]];
                [outputStream close];
            }
        }
            break;
		case NSStreamEventHasBytesAvailable:
        {
			if (stream == inputStream)
			{
				uint8_t buffer[1024];
				int len;
				while ([inputStream hasBytesAvailable])
				{
					len = [inputStream read: buffer maxLength: sizeof(buffer)];
					if (len > 0)
					{
						NSData *theData = [[NSData alloc] initWithBytes: buffer length:len];
                        [self connection: nil didReceiveData: theData];
					}
				}
			}
        }
            break;
    }
}

As you can see, this method kind of delegates the occurring events to calls to the NSURLConnectionDelegate (which the containing class also implements, due to my initial effort to read the changes feed using a normal NSURLConnection).

Then, the connection:didReceiveData method handles the actual changes:

- (void)connection:(NSURLConnection *)connection didReceiveData:(NSData *)data {
    NSString *str = [[NSString alloc] initWithData:data encoding: NSUTF8StringEncoding];

    NSArray *ary = [str componentsSeparatedByString:@"\n"];
    for (NSString *line in ary) {
        if ([line length] > 0 && [line characterAtIndex: 0] == '{') {
            id json = [line JSONValue];

            NSDictionary *dict = (NSDictionary*)json;
            NSObject *seq = [dict objectForKey: @"seq"];

            if (seq && delegate) {
                [delegate changeReceived: dict];
            }
            NSObject *last_seq = [dict objectForKey: @"last_seq"];
            if (last_seq && delegate) {
                int last = [((NSNumber*)last_seq) intValue];
                [delegate lastSequence: last];
            }
        }
    }

}

This method converts the JSON data to a NSDictionary and forwards these changes to a (self-defined) @protocol called ChangesDelegate:

@protocol ChangesDelegate

@optional

-(void) changeReceived:(NSDictionary*) dict;
-(void) lastSequence:(int) last;
-(void) changesComplete:(NSError*)error;

@end

The implementation that performs the actual calls to CouchDB can be created using:

    m_changes = [[Changes alloc] initWithHost: @"localhost" database: @"db" andFilter: @"app/events"];
    m_changes.delegate = self;

    m_lastSeq = [m_changes last_seq];

Since the calling class also implements the ChangesDelegate @protocol, it will receive the changes events. I am currently using it to display MKAnnotations onto a MKMapView and it works like a charm!

Thank you CouchDB for making life easy!

p.s. I’ve made the Changes class available through github: changesfeed_xcode. Enjoy!