anchor.diff:  
   to_binary(Val) when is_binary(Val) -> Val;
   to_binary(Val) when is_list(Val) -> list_to_binary(Val).
  +
  +extract_bucket_and_stream(Topic, Username, Sep) when is_list(Topic) ->
  +    BTopic = list_to_binary(Topic),
  +    extract_bucket_and_stream(BTopic, Username, Sep);
  +extract_bucket_and_stream(Topic, Username, Sep) ->
  +    case binary:split(Topic, [Sep]) of
  +        [Stream] -> {Username, Stream};
  +        [Bucket, Stream] -> {Bucket, Stream};
  +        [Bucket|Tail] -> {Bucket, list_to_binary(Tail)}
  +    end.
  +
  +topic_from_bucket_and_stream(Bucket, Stream, Sep) ->
  +    list_to_binary([Bucket, Sep, Stream]).
  
positive.code:     lager:error("Error ~p", [Error]),
      {ok, State}.
  
  publish(State=#state{username=Username}, {Topic, Qos, Dup, Retain, MessageId, Payload}) ->
      lager:info("Publish ~p ~p ~p ~p ~p ~p",
                 [Topic, Qos, Dup, Retain, MessageId, Payload]),
      iorio:put(Username, to_binary(Topic), Payload),
      {ok, State}.
  
  subscribe(State=#state{username=Username}, Topics) ->
      R = lists:foldl(fun ({TopicName, Qos}, {QosList, IState}) ->
                              OState = IState,
                              QosVal = Qos,
                              lager:info("Subscribe ~p ~p", [TopicName, Qos]),
                              iorio:subscribe(Username, to_binary(TopicName),
                                              self()),
                              {[QosVal|QosList], OState}
                      end, {[], State}, Topics),
      {QoSList, NewState} = R,
      {ok, QoSList, NewState}.
  
  unsubscribe(State=#state{username=Username}, Topics) ->
      NewState = lists:foldl(fun (TopicName, IState) ->
                              OState = IState,
                              lager:info("Unsubscribe ~p", [TopicName]),
                              iorio:unsubscribe(Username, to_binary(TopicName),
                                                self()),
                              OState
                      end, State, Topics),
      {ok, NewState}.
  
  timeout(State=#state{}, InactiveMs) ->
      lager:info("session was inactive for ~p ms", [InactiveMs]),
      {ok, State}.
  
  info(State=#state{send=Send}, {entry, _Bucket, Topic,
                                 #sblob_entry{timestamp=_Timestamp,
                                              seqnum=_SeqNum, data=Data}}) ->
      % XXX: generate a msg id
      MsgId = 1,
      Send(MsgId, Topic, Data),
      {ok, State};
  
negative.code:          disconnect/1, error/2, publish/2, subscribe/2, unsubscribe/2,
           login/2]).
  
  -record(state, {username, send, bucket_separator}).
  
  -include_lib("sblob/include/sblob.hrl").
  
  init(Opts) ->
      {mqttl_send, Send} = proplists:lookup(mqttl_send, Opts),
      BucketSeparator = proplists:get_value(bucket_separator, Opts, <<"!">>),
      DefaultUsername = proplists:get_value(default_username, Opts, <<"mqtt">>),
      {ok, #state{send=Send, bucket_separator=BucketSeparator,
                  username=DefaultUsername}}.
  
  ping(State=#state{}) -> {ok, State}.
  connect(State=#state{}) -> {ok, State}.
  disconnect(State=#state{}) -> {ok, State}.
  