The ruby extension ObjC Lib

rb_pabst.h

#import <ObjFW/ObjFW.h>
#import "RiakProtobuf.h"
#import <stdint.h>
#import "ruby.h"

#define Get_RiakProtobuf(_rpb) ((RiakProtobuf *)DATA_PTR(_rpb))

VALUE rb_list_keys_request(VALUE, VALUE);

rb_pabst.m

#import "rb_pabst.h"

VALUE rb_list_keys_request(VALUE self, VALUE bucket_name) {
  RiakProtobuf   *riakpb  = Get_RiakProtobuf(self);
  OFString       *bucket;
  OFMutableArray *keys;
  VALUE           rb_request;

  // @TODO: Ensure that 'bucket_name' is a String, or call :to_s. Raise intelligent exception if that fails
  // @TODO: Figure out how to get the encoding of the String. Assumption is a UTF-8 encoding, for now.
  bucket      = [OFString stringWithCString:RSTRING_PTR(bucket_name)
                                     length:RSTRING_LEN(bucket_name)];

  keys        = [[riakpb listKeysInBucket:bucket]
                 autorelease];

  rb_request  = rb_ary_new2([keys count]);

  [keys enumerateObjectsUsingBlock:^(id key, size_t index, BOOL *stop) {
    rb_ary_store(rb_request, index, rb_str_new([key cString], [key cStringLength]));
  }];

  return rb_request;
}

rb_riakpb.m

#import "rb_riakpb.h"

VALUE               rb_mRiakpb  = Qnil;
VALUE               rb_cPabst   = Qnil;
OFAutoreleasePool  *pool        = nil;

void Init_Riakpb() {
  rb_mRiakpb  = rb_define_module("Riakpb");
  rb_cPabst   = rb_define_class_under(rb_mRiakpb, "Pabst", rb_cObject);

  rb_define_alloc_func(rb_cPabst, pabst_allocate);

  rb_define_method(rb_cPabst, "initialize",           pabst_initialize,       0);
  rb_define_method(rb_cPabst, "list_keys_request",    rb_list_keys_request,   1);
}

VALUE pabst_allocate(VALUE klass) {
  RiakProtobuf *riakpb;
  
  pool    = [OFAutoreleasePool alloc];
  riakpb  = [RiakProtobuf alloc];

  return Data_Wrap_Struct(klass, pabst_mark, pabst_free, riakpb);
}

VALUE pabst_initialize(VALUE self) {
  RiakProtobuf *riakpb = Get_RiakProtobuf(self);

  [pool init];

  [riakpb initWithService:@"8087" onNode:@"127.0.0.1"];

  return self;
}

void pabst_mark(RiakProtobuf *self) {
  /* Nothing to do, right now */
}

void pabst_free(RiakProtobuf *self) {
  [self release];
  [pool release];
}

RiakProtobuf.mm

//
//  RiakProtobuf.m
//  riak_pb-objc
//
//  Created by Scott Gonyea on 9/1/10.
//  Copyright (c) 2010 Inherently Lame, Inc. All rights reserved.
//

extern "C" {
#   import "RiakProtobuf.h"
}
#import "riakclient.pb.h"

@implementation RiakProtobuf

- (id)init {
  self    = [super init];
  socket  = [OFTCPSocket socket];

  return self;
}

- (id)initWithService:(OFString *)port onNode:(OFString *)node {
  self    = [super init];
  socket  = [OFTCPSocket socket];

  [self connectToService:port onNode:node];

  return self;
}

- (void)connectToService:(OFString *)port onNode:(OFString *)node {
  [socket connectToService:port onNode:node];
  [socket setKeepAlivesEnabled:YES];
}

- (void)sendMessageWithLength:(uint32_t)length message:(char *)message messageCode:(uint8_t)code routeResponse:(BOOL)route {

}

- (void)sendMessageWithLength:(uint32_t)length message:(char *)message messageCode:(uint8_t)code {
  [self sendMessageWithLength:length message:message messageCode:code routeResponse:YES];
}

- (void)dealloc {
  // Clean-up code here.
  [socket release];
  google::protobuf::ShutdownProtobufLibrary();
  [super dealloc];
}

- (OFMutableArray *)listKeysInBucket:(OFString *)bucket {
  RpbListKeysReq pbMsg;
  char     *message;
  uint32_t  msgLen;
  uint8_t   msgCode = MC_LIST_KEYS_REQUEST;

  pbMsg.set_bucket([bucket cString]);

  msgLen  = pbMsg.ByteSize();
  message = (char *)[self allocMemoryWithSize:msgLen];

  // @TODO: Create an Ostream C++ class and Serialize directly onto the socket buffer.
  pbMsg.SerializeToArray(message, msgLen);

  [socket bufferWrites];
  [socket writeBigEndianInt32:(msgLen + MSG_CODE_SIZE)];
  [socket writeInt8:msgCode];
  [socket writeNBytes:msgLen fromBuffer:message];
  [socket flushWriteBuffer];

  msgLen  = [socket readBigEndianInt32] - MSG_CODE_SIZE;
  message = (char *)[self allocMemoryWithSize:msgLen];
  msgCode = [socket readInt8];

  [socket readNBytes:msgLen intoBuffer:message];

  return [self listKeysResponse:message length:msgLen fromBucket:bucket];

//  delete pbMsg;
}

- (OFMutableArray *)listKeysResponse:(char *)response length:(uint32_t)length fromBucket:(OFString *)bucket {
  RpbListKeysResp pbuff;
  BOOL isDone;
  int keyIndex = 0;
  OFMutableArray *keys = [[OFMutableArray array] retain];

  pbuff.ParseFromArray(response, length);
  if(pbuff.has_done()) {
    isDone = pbuff.done();
  }

  for(; keyIndex < pbuff.keys_size(); keyIndex++) {
    [keys addObject:[OFString stringWithCString:pbuff.keys(keyIndex).c_str()]];
  }

  return keys;
}

@end