SocketRocket and PrivatePub
At work we're using Ryan Bates excellent Private Pub gem to push updates from background workers to web clients. Our client wanted an iOS App as well and since I recently read about SocketRocket I was wondering if I could keep the current infrastructure and push data to the iOS App as well.
The web app consists of a Ruby on Rails 3.2 frontend which utilizes Resque for long running background jobs. The results of the Resque jobs are are pushed to all connected clients using Private Pub. Pretty simple. However you need to make some adjustments for SocketRocket and Private Pub to play nice.
TL; DR;
I've uploaded the source code for a sample Ruby on Rails app as well as a sample implementation for iOS on Github. See
You can follow the setup of the Ruby on Rails app using my commits. The Rails App uses the following gems:
private_pub- Ryan Bates Private Pub gemfaye&faye-websocket- we need both to handle websocket clientsthin- executesprivate_pubforeman- to managerails&private_pub
The iOS SocketRocket client depends on:
SocketRocket- websocket client implementation in Objective-CSBJson- JSON parsing framework to process server responsesAFNetworking- to fetch the websocket client configuration
In this post I'll try to outline the basic steps to get you up and running.
Server first - Ruby on Rails configuration & setup
Exposing Private Pub subscription informations
Private Pub protects us from unwanted clients using a signature to authenticate each subscriber. For SocketRocket to talk to Private Pub we need to get the required authentication informations first.
class Api::WebsocketsController < ApplicationController
respond_to :json
def configuration
subscription = PrivatePub.subscription channel: params.fetch(:channel) { '/messages/new' }
respond_with subscription
end
end
WebsocketsController#configuration serves the PrivatePub subscription informations to a potential client. Note that this is a very simple example. I did not add things like API authentication which could be easily done using devise or authorization which could also be done easily using cancan.
Add the following to make the WebsocketsController accessible to the outside:
namespace :api do
match 'websockets/configuration' => 'websockets#configuration', via: :get
end
Testing WebsocketsController#configuration
Once you've started the example application you should be able to test this using curl:
$ curl http://localhost:3000/api/websockets/configuration.json
{"server":"http://localhost:9292/faye","timestamp":1334429946136,"channel":"/messages/new","signature":"f2f893c52621018409acc634ef540971c48347f4"}
Publishing messages from the rails console
To test our SocketRocket client later on we're going to use rails console to publish messages:
PrivatePub.publish_to '/messages/new', message: 'Hello, World!'
That's all on the Rails side! Details can be reviewed on the project side @github.
iOS: SocketRocket client
I'm going to skip the basic setup because there are enough informations available at the respective project sides. I'm assuming you've got an iOS app setup using SBJson, AFNetworking and SocketRocket.
Fetching the Private Pub configuration
First, we need to retrieve the Private Pub subscription informations. This is done using AFNetworking. I've left out the part where we're creating a Private Pub client for now:
NSString *resourceUrl = [NSString stringWithFormat:@"http://localhost:3000/api/websockets/configuration.json?channel=%@", @"/messages/new"];
NSURL *url = [NSURL URLWithString:resourceUrl];
NSURLRequest *request = [NSURLRequest requestWithURL:url];
AFJSONRequestOperation *operation = [AFJSONRequestOperation JSONRequestOperationWithRequest:request
success:^(NSURLRequest *request, NSHTTPURLResponse *response, id JSON) {
[self initializePrivatePubClientWithSubscriptionInformation: JSON];
}
failure:nil
];
[operation start];
After fetching the Private Pub subscription informations I'm passing them along to the initializePrivatePubClientWithSubscriptionInformation method.
Instanciating a new Private Pub client
The initializePrivatePubClientWithSubscriptionInformation method sets up a SRWebSocket client, as well as delegate which takes care of the Private Pub connection setup. Please note that this could be wrapped differently - I'm assuming all we need is a consuming client. If this were to be used for bidirectional activity I'd have to wrap it differently.
- (void) initializePrivatePubClientWithSubscriptionInformation: (id) JSON {
self.websocketDelegate = [[PrivatePubWebSocketDelegate alloc]
initWithPrivatePubTimestamp: [JSON valueForKeyPath:@"timestamp"]
andSignature: [JSON valueForKeyPath:@"signature"]
andChannel:NEW_MESSAGES_CHANNEL];
NSString *server = [JSON valueForKeyPath:@"server"];
NSURL *url = [NSURL URLWithString:server];
NSURLRequest *configurationRequest = [NSURLRequest requestWithURL:url];
self.websocketClient = [[SRWebSocket alloc] initWithURLRequest:configurationRequest];
self.websocketClient.delegate = self.websocketDelegate;
[self.websocketClient open];
}
That's the basic setup. Now let's take a look under the hood to see what PrivatePubWebSocketDelegate has to do for us so we can use Private Pub.
PrivatePubWebSocketDelegate
Before we can receive or publish any messages we need to send two messages to our Private Pub server. Namely:
- a message to
/meta/handshaketo introduce us. - a message to
/meta/subscribeto subscribe to a channel we want to publish/ receive messages from.
For additional logic like connection keep-alive and the client-netiquette we also need to send
- a message to
/meta/disconnectwhen we exit our client - a message to
/meta/connectto ping the server
Since we can neither receive nor publish messages before our subscription was successful I've split the setup logic up using a very simplish state machine. The state machine is implemented using isa-swizzling.
Using a state machine gives me proper separation of concern: every state handles the messages necessary and changes the state as soon as necessary.
I've split up the Private Pub client into the following states:
AwaitingHandshakeStatewhich handles the introduction, including assigning a clientId and advancing us into the...SubscriptionStatewhich subscribes us to a single channel. After that we reach theKeepAliveStatewhich periodically sends a keep-alive message to the server. Typically we'd add our custom logic at this stage.
PrivatePubWebSocketDelegate & AwaitingHandshakeState
Let's take a look at the entry point for our AwaitingHandshakeState:
@interface PrivatePubWebSocketDelegate : NSObject <SRWebSocketDelegate>
// *snip*
@end
@implementation PrivatePubWebSocketDelegate
// *snip*
#pragma mark - SRWebSocketDelegate Protocol Methods
// *snip*
- (void)webSocketDidOpen:(SRWebSocket *)webSocket {
NSLog(@"WebSocket is open");
NSDictionary *handshake = [NSDictionary dictionaryWithObjectsAndKeys:
@"/meta/handshake", @"channel",
[NSNumber numberWithFloat:1.0], @"version",
[NSArray arrayWithObjects:@"websocket", nil], @"supportedConnectionTypes",
[NSNumber numberWithInt:self.messageId++], @"id",
nil];
isa = [AwaitingHandshakeState class];
[webSocket send:[handshake JSONRepresentation]];
}
@end
The PrivatePubWebSocketDelegate base class is a simple SRWebSocketDelegate. It's "kind of" a state machine because
it changes it's own class once the connection has been opened. After the connection has been opened we have to tell Private Pub
what kind of client we are. This includes a list of supported connection types:
NSDictionary *handshake = [NSDictionary dictionaryWithObjectsAndKeys:
@"/meta/handshake", @"channel",
[NSNumber numberWithFloat:1.0], @"version",
[NSArray arrayWithObjects:@"websocket", nil], @"supportedConnectionTypes",
[NSNumber numberWithInt:self.messageId++], @"id",
nil];
// *snip*
[webSocket send:[handshake JSONRepresentation]];
The response tells us if our handshake was successful or not. This logic is handled in the AwaitingHandshakeState:
@implementation AwaitingHandshakeState
- (void) webSocket:(SRWebSocket *)webSocket didReceiveMessage:(NSString *)message {
id JSON = [[message JSONValue] objectAtIndex:0];
NSString *channel = [JSON valueForKeyPath:@"channel"];
if ([channel isEqualToString:@"/meta/handshake"]) {
Boolean handshakeWasSuccessful = [[JSON valueForKeyPath:@"successful"] boolValue];
if (handshakeWasSuccessful) {
self.clientId = [JSON valueForKeyPath:@"clientId"];
self.webSocket = webSocket;
isa = [SubscriptionState class];
[(SubscriptionState *)self sendSubscriptions];
}
}
}
@end
First, I'm making sure our received message is an actual response to our handshake: [channel isEqualToString:@"/meta/handshake"].
After that I'm inspecting the message if the handshake was successful: [[JSON valueForKeyPath:@"successful"] boolValue].
If this is the case the clientId is set, and I'm advancing the state machine into the SubscriptionState. Note that the clientId must be present in all
messages following the handshake, and we can't choose our own. It's assigned by the server.
the SubscriptionState
The SubscriptionState takes care of sending messages to /meta/subscribe to tell Private Pub we are actually interested in a certain channel.
Also it uses the passed in signature and timestamp to pass Private Pub validation. Since we have to take the initiative on this, AwaitingHandshakeState
tells us to sendSubscriptions:
@implementation SubscriptionState
- (void) sendSubscriptions {
NSDictionary *ext = [NSDictionary dictionaryWithObjects:[NSArray arrayWithObjects:self.privatePubTimestamp, self.privatePubSignature, nil] forKeys:[NSArray arrayWithObjects:@"private_pub_timestamp", @"private_pub_signature", nil]];
NSDictionary *subscription = [NSDictionary dictionaryWithObjectsAndKeys:
@"/meta/subscribe", @"channel",
self.clientId, @"clientId",
self.channel, @"subscription",
[NSNumber numberWithInt:self.messageId++], @"id",
ext, @"ext",
nil];
[self.webSocket send:[subscription JSONRepresentation] ];
}
- (void) webSocket:(SRWebSocket *)webSocket didReceiveMessage:(NSString *)message {
id JSON = [[message JSONValue] objectAtIndex:0];
NSLog(@"recv subscription %@", JSON);
Boolean subscriptionWasSuccessful = [[JSON valueForKeyPath:@"successful"] boolValue];
if (subscriptionWasSuccessful) {
NSLog(@"Now subscribed to %@", self.channel);
isa = [KeepAliveState class];
[(KeepAliveState *)self setupKeepAlive];
}
}
@end
The sendSubscriptions method takes care of publishing our subscription request to /meta/subscribe, while our new websocket:didReceiveMessage: method
checks if the subscription was actually successful. Only if the subscription was successfully established we're going to enter our last state: the KeepAliveState.
keeping the connection alive: KeepAliveState
The setupKeepAlive method detaches a new thread which sends a single keep-alive message. I'm assuming that we want to keep the connection open forever.
In this case the sendKeepAlive method detaches itself to send another keep-alive message.
@implementation KeepAliveState
- (void) setupKeepAlive {
// *snip*
[NSThread detachNewThreadSelector:@selector(sendKeepAlive) toTarget:self withObject:nil];
// *snip*
}
- (void) sendKeepAlive {
sleep(self.timeout);
NSDictionary *keepAlive = [NSDictionary dictionaryWithObjectsAndKeys:
self.clientId, @"clientId",
@"/meta/connect", @"channel",
@"websocket", @"connectionType",
[NSNumber numberWithInt:self.messageId++], @"id",
nil];
[self.webSocket send: [keepAlive JSONRepresentation]];
[NSThread detachNewThreadSelector:@selector(sendKeepAlive) toTarget:self withObject:nil];
}
@end
Note that I did not implement a proper NSThread handling here. It's probably a very bad example - just for demonstration purposes.
Anyway, that's about it. Lets wrap this up:
- Our state machine using inheritance with
PrivatePubWebSocketDelegateas superclass - every state handles different messages, so every subclass has its own
webSocket:didReceiveMessage:method which handles only specific messages - the
AwaitingHandshakeStateis reached first. If the handshake was successful it assigns a clientId - the
SubscriptionStateis reached next. It publishes a subscription request with the Private Pub authentication informations and checks if the request was successful. - the
KeepAliveStateis reached last. It usesNSThreadto publish a keep-alive message everytimeoutseconds. This is were we'd want to add our custom Private Pub client-server logic.
I guess this could be properly re-factored and released as a separate library, since I have not found any resources on how to write a SocketRocket - Private Pub client so far. Feel free to contact me if you are having troubles reproducing this. Also note that you can always try out the sample projects @github.
Happy hacking!
Links
- bayeux message format documentation This helped me a lot when I was stuck with Private Pub error messages.