Commits

Anonymous committed 7167f5c

initial commit of basically end-to-end working Wriaki

  • Participants
  • Parent commits 6795e94

Comments (0)

Files changed (53)

 \.beam$
 ^rel/wriaki
 ^deps/.*
+~$
 
+.PHONY: rel
+
 all:
 	@./rebar compile
 
+rel: all
+	@./rebar generate
+
+relforce: all
+	@./rebar generate force=1
+Wriaki: the Riak-based Wiki
+
+* Overview
+
+Wriaki is a wiki-like web application, intended to illustrate a few
+strategies for storing data in Riak.
+
+* Installation
+
+** Prerequisites
+
+To run Wriaki, you will need Riak, Python and the Wiki Creole Python
+package.
+
+*** Riak
+
+The easiest way to get Riak is to download a pre-built distribution
+from [http://downloads.basho.com/riak/].  Any version 0.9.1 or newer
+should work.
+
+Riak must be configured to expose its HTTP interface.  By default,
+Riak does this on localhost port 8098.  If you'll be running Wriaki on
+the same machine, you can just leave this setting at its default.
+
+*** Wiki Creole
+
+The easiest way to get Wiki Creole is by using easy_install:
+
+: $ easy_install wiki_creole
+
+** Downloading and Building Wriaki
+
+To setup Wriaki, first clone the source:
+
+: $ hg clone http://bitbucket.org/basho/wriaki
+
+Next, change to the source directory and run make:
+
+: $ cd wriaki
+: $ make rel
+
+** Configuration
+
+After building, you should have a "rel/wriaki/" subdirectory under the
+source directory.  Configuration for wriaki is stored in
+"rel/wriaki/etc/app.config".
+
+The settings that Wriaki knows about are:
+
+ + salt :: the "salt" used for encrypting user passwords
+
+ + riak_ip :: the IP address of the machine in the Riak cluster to
+              connect to
+
+ + riak_port :: the TCP port the Riak node is listening on
+
+ + riak_prefix :: the URL prefix for Riak data
+
+                  http://<riak_ip>:<riak_port>/<riak_prefix>/Bucket/Key
+
+ + web_ip :: the IP to bind Wriaki's webserver to
+
+ + web_port :: the TCP port Wriaki should listen on
+
+ + log_dir :: the directory to write Wriaki's access log in
+
+* Running
+
+Before running Wriaki, ensure that your Riak cluster is started and
+reachable.
+
+Next, run the wriaki script in your rel/wriaki/bin/ subdirectory:
+
+: $ rel/wriaki/bin/wriaki console
+
+To start Wriaki in the background, use "start" instead of "console" on
+that command line.
+
+* Data Layout
+
+There are four basic objects in the Wriaki system: article, archive,
+history, and user.
+
+** Article
+
+One 'article' object exists for each page on the wiki.
+
+*** Key: article title
+
+The key for an article object is the title of the wiki page,
+url-encoded.
+
+*** Bucket: article
+
+Articles are stored in the 'article' Riak bucket.  The 'article'
+bucket is configured for 'allow_mult=true'.  This is done to allow
+multiple users to edit an article concurrently.  If they save at the
+"same" time, the article object will contain siblings on the next
+read, and Wriaki will warn the viewer that there are multiple versions
+of the article that are currently considered "the latest."
+
+*** Body: json
+
+The value of an article object is JSON, with the fields:
+ + text :: (string) content in wiki markup format
+ + message :: (string) commit message
+ + version :: (string) version hash
+ + timestamp :: (int) edit date
+
+*** Headers
+
+Articles use one link to track which user created that version of the
+object.  The link will be to an object in the 'user' bucket, and will
+be tagged 'editor'.
+
+*** Merge: ask user
+
+When conflicting writes to an article are found, the user will be
+given the option to view the version they want.  Editing the article
+will resolve the conflict.
+
+** Archive
+
+One archive object exists for each version (past and present) of each
+article.
+
+*** Key: version.article
+
+The key for an archive object is the version hash appended with the
+article object key, separated by a dot.
+
+*** Bucket: archive
+
+Archive objects are stored in the 'archive' bucket.  The bucket is
+left as 'allow_mult=false'.
+
+*** Body: json
+
+The value of an archive object is exactly the same as that of an
+article object.
+
+*** Headers
+
+The archive object has the same link header as the article object.
+
+*** Merge: last write wins
+
+Archive objects should be write-once, due to their key generation, and
+thus will not need a merge strategy.
+
+** History
+
+One history object exists for each page on the wiki.  The purpose of
+the history object is to hold links to all versions of each article
+object.
+
+*** Key: article
+
+The key for the history object is the same as the key for the article
+object.
+
+*** Bucket: history
+
+History objects are stored in the 'history' bucket.  The bucket is
+configured for 'allow_mult=true' to allow multiple users to add
+article versions (thus updating the history) concurrently.
+
+*** Body: empty
+
+History objects have no data in their bodies.
+
+*** Headers
+
+History object have one link for each version an article has had.  The
+links will target objects in the 'archive' bucket, and will be tagged
+with the timestamp of the article version.
+
+*** Merge: set-union links
+
+Merging two versions of an archive object is simply set-unioning the
+list of links.
+
+** User
+
+One user object exists for each registered user of the wiki.  This
+object keeps track of the user's password and other data.
+
+*** Key: username
+
+User objects are keyed by url-encoded usernames.
+
+*** Bucket: user
+
+User objects are stored in the 'user' bucket.  The bucket is left as
+'allow_mult=false' because only the user should be updating that
+user's object (no concurrent writing).
+
+*** Body: json
+
+The value of a user object is JSON with the fields:
+
+ + email :: (string) email address
+ + password :: (string, base64) encrypted
+ + bio :: (string) short biography
+
+*** Headers
+
+User object have no headers.
+
+*** Merge: last write wins
+
+No merge is needed for user objects.  They should only be edited by
+their owners, and last-write-wins will be good enough to handle that.
+
+* Web Resources
+
+Wriaki exposes the following resources:
+
+ + /user :: login page, GET-only
+ + /user/<username> :: User's settings
+
+      GET: with no query parameters returns a page of public
+           information about the user
+           
+           with query parameter ?edit, returns a form for the user to
+           update their information (user is redirected to
+           non-query-parameter URL if this is not their login)
+
+      PUT: change user data
+
+      POST: login
+
+ + /user/<username>/<sessionid> :: Session information
+
+      GET: get expiry time of the session, also extends the session's
+           expiry
+
+      DELETE: remove the session, "logout"
+
+ + /wiki/<page name> :: Wiki page
+
+      GET: with no query parameters returns the rendered wiki page
+
+           with query parameter ?edit, returns a form for the user to
+           edit the page
+
+           with query parameter ?history, returns a list of the known
+           versions of the object
+
+           with query parameter ?v=<version>, returns the page
+           rendered for the requested version
+
+           with query paramaters
+           ?diff&l=<left_version>&r=<right_version> returns a
+           line-by-line difference of the given versions
+
+      PUT: store a new version of the wiki page
+
+      POST: preview a new version of the wiki page
+
+ + /static/* :: serve static files from disk
+
+      GET: retrieve the specified file

File apps/riak_erlang_client/ebin/riak_erlang_client.app

+{application, riak_erlang_client,
+ [
+  {description, ""},
+  {vsn, "1"},
+  {modules, [
+             riak_erlang_client_app,
+             riak_erlang_client_sup,
+             rhc,
+             rec_obj
+            ]},
+  {registered, []},
+  {applications, [
+                  kernel,
+                  stdlib
+                 ]},
+  {mod, { riak_erlang_client_app, []}},
+  {env, []}
+ ]}.

File apps/riak_erlang_client/include/raw_http.hrl

+%% This file is provided to you under the Apache License,
+%% Version 2.0 (the "License"); you may not use this file
+%% except in compliance with the License.  You may obtain
+%% a copy of the License at
+
+%%   http://www.apache.org/licenses/LICENSE-2.0
+
+%% Unless required by applicable law or agreed to in writing,
+%% software distributed under the License is distributed on an
+%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+%% KIND, either express or implied.  See the License for the
+%% specific language governing permissions and limitations
+%% under the License.    
+
+%% Constants used by the raw_http resources
+%% original source at
+%% http://bitbucket.org/basho/riak/src/tip/apps/riak/src/raw_http.hrl
+
+%% Names of riak_object metadata fields
+-define(MD_CTYPE,    <<"content-type">>).
+-define(MD_CHARSET,  <<"charset">>).
+-define(MD_ENCODING, <<"content-encoding">>).
+-define(MD_VTAG,     <<"X-Riak-VTag">>).
+-define(MD_LINKS,    <<"Links">>).
+-define(MD_LASTMOD,  <<"X-Riak-Last-Modified">>).
+-define(MD_USERMETA, <<"X-Riak-Meta">>).
+
+%% Names of HTTP header fields
+-define(HEAD_CTYPE,           "Content-Type").
+-define(HEAD_VCLOCK,          "X-Riak-Vclock").
+-define(HEAD_LINK,            "Link").
+-define(HEAD_ENCODING,        "Content-Encoding").
+-define(HEAD_CLIENT,          "X-Riak-ClientId").
+-define(HEAD_USERMETA_PREFIX, "x-riak-meta-").
+
+%% Names of JSON fields in bucket properties
+-define(JSON_PROPS,   <<"props">>).
+-define(JSON_KEYS,    <<"keys">>).
+-define(JSON_LINKFUN, <<"linkfun">>).
+-define(JSON_MOD,     <<"mod">>).
+-define(JSON_FUN,     <<"fun">>).
+-define(JSON_CHASH,   <<"chash_keyfun">>).
+-define(JSON_JSFUN,    <<"jsfun">>).
+-define(JSON_JSANON,   <<"jsanon">>).
+-define(JSON_JSBUCKET, <<"bucket">>).
+-define(JSON_JSKEY,    <<"key">>).
+-define(JSON_ALLOW_MULT, <<"allow_mult">>).
+
+
+%% Names of HTTP query parameters
+-define(Q_PROPS, "props").
+-define(Q_KEYS,  "keys").
+-define(Q_FALSE, "false").
+-define(Q_TRUE, "true").
+-define(Q_STREAM, "stream").
+-define(Q_VTAG,  "vtag").
+-define(Q_RETURNBODY, "returnbody").

File apps/riak_erlang_client/src/rec_obj.erl

+-module(rec_obj).
+
+-export([create/3,
+         create_siblings/4,
+         get_vclock/1,
+         set_vclock/2,
+         bucket/1,
+         key/1,
+         get_value/1,
+         set_value/2,
+         get_links/1,
+         set_links/2,
+         remove_links/3,
+         add_link/2,
+         get_content_type/1,
+         set_content_type/2,
+         get_siblings/1,
+         has_siblings/1,
+         get_json_field/2,
+         set_json_field/3]).
+         
+-record(rec_obj, {bucket,
+                  key,
+                  value,
+                  links=[],
+                  ctype="application/json",
+                  vclock=""}).
+-record(rec_sib, {bucket,
+                  key,
+                  vclock,
+                  sibs}).
+
+create(Bucket, Key, Value) when is_binary(Bucket), is_binary(Key) ->
+    #rec_obj{bucket=Bucket, key=Key, value=Value}.
+
+create_siblings(Bucket, Key, Vclock, Siblings)
+  when is_binary(Bucket), is_binary(Key),
+       is_list(Vclock), is_list(Siblings) ->
+    #rec_sib{bucket=Bucket, key=Key, vclock=Vclock, sibs=Siblings}.
+
+get_vclock(#rec_obj{vclock=Vclock}) ->
+    Vclock;
+get_vclock(#rec_sib{vclock=Vclock}) ->
+    Vclock.
+
+set_vclock(Obj, Vclock) ->
+    Obj#rec_obj{vclock=Vclock}.
+
+bucket(#rec_obj{bucket=Bucket}) ->
+    Bucket;
+bucket(#rec_sib{bucket=Bucket}) ->
+    Bucket.
+
+key(#rec_obj{key=Key}) ->
+    Key;
+key(#rec_sib{key=Key}) ->
+    Key.
+
+get_value(#rec_obj{value=Value}) ->
+    Value.
+
+set_value(Obj, Value) ->
+    Obj#rec_obj{value=Value}.
+
+get_links(#rec_obj{links=Links}) ->
+    Links.
+
+remove_links(Obj, Bucket, Tag) ->
+    set_links(Obj, lists:filter(link_filter(Bucket, Tag), get_links(Obj))).
+
+link_filter('_', '_') ->
+    fun(_) -> true end;
+link_filter(Bucket, '_') ->
+    fun({{B,_},_}) -> B =:= Bucket end;
+link_filter('_', Tag) ->
+    fun({_,T}) -> T =:= Tag end;
+link_filter(Bucket, Tag) ->
+    fun({{B,_},T}) -> B =:= Bucket andalso T =:= Tag end.
+
+set_links(Obj, Links) when is_list(Links) ->
+    Obj#rec_obj{links=Links}.
+
+add_link(Obj, Link={{_,_},_}) ->
+    Links = get_links(Obj),
+    case lists:member(Link, Links) of
+        true -> Obj;
+        false -> set_links(Obj, [Link|Links])
+    end.
+
+get_content_type(#rec_obj{ctype=ContentType}) ->
+    ContentType.
+
+set_content_type(Obj, ContentType) ->
+    Obj#rec_obj{ctype=ContentType}.
+
+get_siblings(#rec_sib{bucket=Bucket, key=Key, vclock=Vclock, sibs=Siblings}) ->
+    [set_links(
+       set_content_type(
+         set_vclock(
+           create(Bucket, Key, Value),
+           Vclock),
+         ContentType),
+       Links)
+     || {ContentType, Links, Value} <- Siblings].
+
+has_siblings(#rec_sib{}) -> true;
+has_siblings(#rec_obj{}) -> false.
+
+
+get_json_field(RecObj, Field) ->
+    {struct, Props} = rec_obj:get_value(RecObj),
+    proplists:get_value(Field, Props).
+
+set_json_field(RecObj, Field, Value) ->
+    {struct, Props} = rec_obj:get_value(RecObj),
+    NewProps = [{Field, Value}
+                | [ {F, V} || {F, V} <- Props, F =/= Field ]],
+    rec_obj:set_value(RecObj, {struct, NewProps}).
+    

File apps/riak_erlang_client/src/rhc.erl

+-module(rhc).
+
+-export([create/0, create/4,
+         prefix/1,
+         get/3, get/4,
+         put/2, put/3,
+         delete/3, delete/4,
+         walk/4,
+         mapred/3,
+         set_bucket/3
+        ]).
+
+-include_lib("raw_http.hrl").
+
+-record(rhc, {ip,
+              port,
+              prefix,
+              options}).
+
+prefix(#rhc{prefix=Prefix}) -> Prefix.
+
+create() ->
+    create("127.0.0.1", 8098, "riak", []).
+
+create(IP, Port, Prefix, Opts0) ->
+    Opts = case proplists:lookup(client_id, Opts0) of
+               none -> [{client_id, random_client_id()}|Opts0];
+               _    -> Opts0
+           end,
+    #rhc{ip=IP, port=Port, prefix=Prefix, options=Opts}.
+
+get(Client, Bucket, Key) ->
+    get(Client, Bucket, Key, []).
+get(Client, Bucket, Key, Options) ->
+    Url = make_url(Client, Bucket, Key, Options),
+    case request(get, Url, ["200", "300"]) of
+        {ok, _Status, Headers, Body} ->
+            {ok, make_rec_obj(Bucket, Key, Headers, Body)};
+        {error, {ok, "404", _, _}} ->
+            {error, notfound};
+        {error, Error} ->
+            {error, Error}
+    end.
+
+put(Client, Object) ->
+    put(Client, Object, []).
+put(Client, Object, Options) ->
+    case rec_obj:has_siblings(Object) of
+        false -> ok;
+        true -> throw(cannot_store_siblings)
+    end,
+    Bucket = rec_obj:bucket(Object),
+    Key = rec_obj:key(Object),
+    Url = make_url(Client, Bucket, Key, Options),
+    Method = if Key =:= undefined -> post;
+                true              -> put
+             end,
+    {Headers0, Body} = serialize_rec_obj(Client, Object),
+    Headers = [{?HEAD_CLIENT, option(client_id, Client, Options)}
+               |Headers0],
+    case request(Method, Url, ["200", "204", "300"], Headers, Body) of
+        {ok, Status, ReplyHeaders, ReplyBody} ->
+            if Status =:= "204" ->
+                    ok;
+               true ->
+                    {ok, make_rec_obj(Bucket, Key, ReplyHeaders, ReplyBody)}
+            end;
+        {error, Error} ->
+            {error, Error}
+    end.
+
+delete(Client, Bucket, Key) ->
+    delete(Client, Bucket, Key, []).
+delete(Client, Bucket, Key, Options) ->
+    Url = make_url(Client, Bucket, Key, Options),
+    Headers = [{?HEAD_CLIENT, option(client_id, Client, Options)}],
+    case request(delete, Url, ["204"], Headers) of
+        {ok, "204", _Headers, _Body} -> ok;
+        {error, Error}               -> {error, Error}
+    end.
+
+walk(Client, Bucket, Key, Spec) ->
+    throw(not_implemented).
+
+mapred(Client, Inputs, Query) ->
+    Body = mochijson2:encode(
+             {struct, [{<<"inputs">>, mapred_encode_inputs(Inputs)},
+                       {<<"query">>, mapred_encode_query(Query)}]}),
+    Headers = [{"Content-Type", "application/json"},
+               {"Accept", "application/json"}],
+    Url = mapred_url(Client),
+    case request(post, Url, ["200"], Headers, Body) of
+        {ok, "200", ReplyHeaders, ReplyBody} ->
+            {ok, mochijson2:decode(ReplyBody)};
+        {error, Error} ->
+            {error, Error}
+    end.
+
+set_bucket(Client, Bucket, Props) ->
+    Url = make_url(Client, Bucket, undefined, []),
+    Headers =  [{"Content-Type", "application/json"}],
+    Body = mochijson2:encode({struct, [{<<"props">>, {struct, Props}}]}),
+    case request(put, Url, ["204"], Headers, Body) of
+        {ok, "204", _Headers, _Body} -> ok;
+        {error, Error}               -> {error, Error}
+    end.
+
+%% INTERNAL
+
+root_url(#rhc{ip=Ip, port=Port}) ->
+    ["http://",Ip,":",integer_to_list(Port),"/"].
+
+mapred_url(Client) ->
+    binary_to_list(iolist_to_binary([root_url(Client), "mapred/"])).
+
+make_url(Client=#rhc{prefix=Prefix}, Bucket, Key, Options) ->
+    binary_to_list(
+      iolist_to_binary(
+        [root_url(Client), Prefix, "/",
+         Bucket,"/",
+         if Key =/= undefined -> [Key,"/"];
+            true              -> []
+         end,
+         option(extra_rul, Client, Options, []),
+         "?",
+         qparam(Client, r), qparam(Client, w),
+         qparam(Client, dw), qparam(Client, rw),
+         qparam(Client, returnbody)])).
+
+qparam(Client, P) ->
+    case option(P, Client) of
+        undefined -> [];
+        Val ->
+            [atom_to_list(P),"=",
+             if is_integer(Val) -> integer_to_list(Val);
+                is_atom(Val)    -> atom_to_list(Val)
+             end]
+    end.
+            
+option(O, Client) ->
+    option(O, Client, [], undefined).
+option(O, Client, Opts) ->
+    option(O, Client, Opts, undefined).
+option(O, #rhc{options=Options}, Opts, Default) ->
+    case proplists:lookup(O, Opts) of
+        {O, Val} -> Val;
+        _ ->
+            case proplists:lookup(O, Options) of
+                {O, Val} -> Val;
+                _        -> get_app_env(O, Default)
+            end
+    end.
+
+get_app_env(Env, Default) ->
+    case application:get_env(wriaki, Env) of
+        {ok, Val} -> Val;
+        undefined -> Default
+    end.
+
+random_client_id() ->
+    {{Y,Mo,D},{H,Mi,S}} = erlang:universaltime(),
+    {_,_,NowPart} = now(),
+    Id = erlang:phash2([Y,Mo,D,H,Mi,S,node(),NowPart]),
+    base64:encode_to_string(<<Id:32>>).
+
+request(Method, Url, Expect) ->
+    request(Method, Url, Expect, [], []).
+request(Method, Url, Expect, Headers) ->
+    request(Method, Url, Expect, Headers, []).
+request(Method, Url, Expect, Headers, Body) ->
+    Accept = {"Accept", "multipart/mixed, */*;q=0.9"},
+    case ibrowse:send_req(Url, [Accept|Headers], Method, Body) of
+        Resp={ok, Status, _, _} ->
+            case lists:member(Status, Expect) of
+                true -> Resp;
+                false -> {error, Resp}
+            end;
+        Error ->
+            Error
+    end.
+
+
+make_rec_obj(Bucket, Key, Headers, Body) ->
+    Vclock = proplists:get_value(?HEAD_VCLOCK, Headers, ""),
+    case ctype_from_headers(Headers) of
+        {"multipart/mixed", Args} ->
+            {"boundary", Boundary} = proplists:lookup("boundary", Args),
+            rec_obj:create_siblings(
+              Bucket, Key, Vclock,
+              decode_siblings(Boundary, Body));
+        {CType, _} ->
+            {_, Links, Value} =
+                decode_content(Headers, Body),
+            rec_obj:set_links(
+              rec_obj:set_content_type(
+                rec_obj:set_vclock(
+                  rec_obj:create(Bucket, Key, Value),
+                  Vclock),
+                CType),
+              Links)
+    end.
+
+ctype_from_headers(Headers) ->
+    mochiweb_util:parse_header(
+      proplists:get_value(?HEAD_CTYPE, Headers)).
+
+decode_siblings(Boundary, "\r\n"++SibBody) ->
+    decode_siblings(Boundary, SibBody);
+decode_siblings(Boundary, SibBody) ->
+    Parts = webmachine_multipart:get_all_parts(
+              list_to_binary(SibBody), Boundary),
+    [ decode_content([ {binary_to_list(H), binary_to_list(V)}
+                       || {H, V} <- Headers ],
+                     binary_to_list(Body))
+      || {_, {_, Headers}, Body} <- Parts ].
+
+decode_content(Headers, Body) ->
+    Links = extract_links(Headers),
+    case ctype_from_headers(Headers) of
+        {"application/json",_} ->
+            {"application/json", Links, mochijson2:decode(Body)};
+        {Ctype, _} ->
+            {Ctype, Links, Body}
+    end.
+
+extract_links(Headers) ->
+    {ok, Re} = re:compile("</[^/]+/([^/]+)/([^/]+)>; *riaktag=\"(.*)\""),
+    Extractor = fun(L, Acc) ->
+                        case re:run(L, Re, [{capture,[1,2,3],binary}]) of
+                            {match, [Bucket, Key,Tag]} ->
+                                [{{Bucket,Key},Tag}|Acc];
+                            nomatch ->
+                                Acc
+                        end
+                end,
+    LinkHeader = proplists:get_value(?HEAD_LINK, Headers, []),
+    lists:foldl(Extractor, [], string:tokens(LinkHeader, ",")).
+
+serialize_rec_obj(Client, Object) ->
+    {make_headers(Client, Object), make_body(Object)}.
+
+make_headers(Client, Object) ->
+    [{?HEAD_LINK, encode_links(Client, rec_obj:get_links(Object))},
+     {?HEAD_CTYPE, rec_obj:get_content_type(Object)}
+     |case rec_obj:get_vclock(Object) of
+          ""     -> [];
+          Vclock -> [{?HEAD_VCLOCK, Vclock}]
+      end].
+
+encode_links(_, []) -> [];
+encode_links(#rhc{prefix=Prefix}, Links) ->
+    {{FirstBucket, FirstKey}, FirstTag} = hd(Links),
+    lists:foldl(
+      fun({{Bucket, Key}, Tag}, Acc) ->
+              [format_link(Prefix, Bucket, Key, Tag), ", "|Acc]
+      end,
+      format_link(Prefix, FirstBucket, FirstKey, FirstTag),
+      tl(Links)).
+
+format_link(Prefix, Bucket, Key, Tag) ->
+    io_lib:format("</~s/~s/~s>; riaktag=\"~s\"",
+                  [Prefix, Bucket, Key, Tag]).
+
+make_body(Object) ->
+    case rec_obj:get_content_type(Object) of
+        "application/json" ->
+            mochijson2:encode(rec_obj:get_value(Object));
+        _ ->
+            rec_obj:get_value(Object)
+    end.
+
+mapred_encode_inputs(Inputs) when is_binary(Inputs) ->
+    Inputs;
+mapred_encode_inputs(Inputs) when is_list(Inputs) ->
+    lists:map(
+      fun({{Bucket, Key}, KeyData}) -> [Bucket, Key, KeyData];
+         ([Bucket, Key, KeyData])   -> [Bucket, Key, KeyData];
+         ({Bucket, Key}) -> [Bucket, Key];
+         ([Bucket, Key]) -> [Bucket, Key]
+      end,
+      Inputs).
+
+mapred_encode_query(Query) ->
+    lists:map(
+      fun({link, Bucket, Tag, Keep}) when is_boolean(Keep) ->
+              {struct,
+               [{<<"link">>,
+                 {struct, [{<<"bucket">>,
+                            if is_binary(Bucket) -> Bucket;
+                               Bucket =:= '_'    -> <<"_">>
+                                                        end},
+                           {<<"tag">>,
+                            if is_binary(Tag) -> Tag;
+                               Tag =:= '_'    -> <<"_">>
+                                                     end},
+                           {<<"keep">>, Keep}]}}]};
+         ({map, {jsanon, Source}, Arg, Keep}) when is_boolean(Keep) ->
+              {struct,
+               [{<<"map">>,
+                 {struct, [{<<"language">>, <<"javascript">>},
+                           {<<"source">>, Source},
+                           {<<"arg">>, Arg},
+                           {<<"keep">>, Keep}]}}]};
+         ({reduce, {jsanon, Source}, Arg, Keep}) when is_boolean(Keep) ->
+              {struct,
+               [{<<"reduce">>,
+                 {struct, [{<<"language">>, <<"javascript">>},
+                           {<<"source">>, Source},
+                           {<<"arg">>, Arg},
+                           {<<"keep">>, Keep}]}}]}
+      end,
+      Query).
+
+                                       
+              
+    

File apps/riak_erlang_client/src/riak_erlang_client_app.erl

+-module(riak_erlang_client_app).
+
+-behaviour(application).
+
+%% Application callbacks
+-export([start/2, stop/1]).
+
+%% ===================================================================
+%% Application callbacks
+%% ===================================================================
+
+start(_StartType, _StartArgs) ->
+    riak_erlang_client_sup:start_link().
+
+stop(_State) ->
+    ok.

File apps/riak_erlang_client/src/riak_erlang_client_sup.erl

+
+-module(riak_erlang_client_sup).
+
+-behaviour(supervisor).
+
+%% API
+-export([start_link/0]).
+
+%% Supervisor callbacks
+-export([init/1]).
+
+%% Helper macro for declaring children of supervisor
+-define(CHILD(I, Type), {I, {I, start_link, []}, permanent, 5000, Type, [I]}).
+
+%% ===================================================================
+%% API functions
+%% ===================================================================
+
+start_link() ->
+    supervisor:start_link({local, ?MODULE}, ?MODULE, []).
+
+%% ===================================================================
+%% Supervisor callbacks
+%% ===================================================================
+
+init([]) ->
+    {ok, { {one_for_one, 5, 10}, []} }.
+

File apps/wiki_creole/ebin/wiki_creole.app

+{application, wiki_creole,
+ [
+  {description, ""},
+  {vsn, "1"},
+  {modules, [
+             wiki_creole_app,
+             wiki_creole_sup,
+             creole,
+             script_manager,
+             script_worker
+            ]},
+  {registered, []},
+  {applications, [
+                  kernel,
+                  stdlib
+                 ]},
+  {mod, { wiki_creole_app, []}},
+  {env, []}
+ ]}.

File apps/wiki_creole/priv/creole_worker.py

+#!/usr/bin/env python
+
+import sys
+import creoleparser
+
+COMMAND_BREAK = "------wriaki-creole-break------"
+Acc = ""
+
+while 1:
+    L = sys.stdin.readline()
+    if L.strip() == COMMAND_BREAK:
+        H = creoleparser.text2html(Acc)
+        print H
+        print COMMAND_BREAK
+        sys.stdout.flush()
+        Acc = ""
+    elif L == "":
+        break
+    else:
+        Acc += L
+

File apps/wiki_creole/rebar.config

+{erl_first_files, ["src/script_worker.erl"]}.
+

File apps/wiki_creole/src/creole.erl

+-module(creole).
+-behaviour(script_worker).
+
+%% user API
+-export([text2html/1]).
+
+%% script_worker API
+-export([init_trigger/0,
+         handle_init/1,
+         process/2,
+         handle_data/2]).
+
+%% user API
+
+%% @spec text2html(iolist()) -> iolist()
+%% @doc Compile Wiki-creole-syntax text into HTML.
+text2html(Text) ->
+    script_manager:process(creole, Text).
+
+%% script_worker API
+
+-define(COMMAND_BREAK, "------wriaki-creole-break------").
+
+%% @private
+init_trigger() -> none.
+
+%% @private
+handle_init(_) -> exit("creole does not use handle_init").
+    
+%% @private
+process(Port, Text) ->
+    port_command(Port, Text),
+    port_command(Port, ["\n", ?COMMAND_BREAK, "\n"]),
+    [].
+
+%% @private
+handle_data(RespAcc, ?COMMAND_BREAK) ->
+    {done, lists:flatten(lists:reverse(RespAcc))};
+handle_data(RespAcc, Line) ->
+    {continue, ["\n",Line|RespAcc]}.

File apps/wiki_creole/src/script_manager.erl

+%% @author Bryan Fink <bryan@basho.com>
+%% @since 8.Apr.2009
+%% @doc Generic server for parallelizing requests to os-processes.
+%%
+%%  This module attempts to solve the problem of distributing requests
+%%  to long-lived stdio-oriented OS-processes.  That is, if you have a
+%%  program (shell script, etc.) that you need to communicate with,
+%%  which reads data from stdin and returns data on stdout, using this
+%%  module (along with {@link script_worker}) should make the
+%%  implementation of this communication simple.  Once that
+%%  implementation is complete, it should also be trivial to move to
+%%  using a pool of these scripts as parallel workers.
+%%
+%%  Implement your script interface according to the instructions in
+%%  {@link script_worker}.
+%%
+%%  Start the manager and workers by calling {@link start_link/4}.
+%%
+%%  Submit work by calling {@link process/2}.
+%%
+%%  If things get confusing, send the message `dump_state' to your
+%%  manager process to get it to print out (via {@link error_logger})
+%%  its current state.
+-module(script_manager).
+
+-behaviour(gen_server).
+
+%% gen_server API
+-export([start_link/4]).
+%% script_worker API
+-export([worker_available/2]).
+%% client API
+-export([process/2]).
+-export([inc_workers/1, dec_workers/1]).
+
+%% gen_server callbacks
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
+         terminate/2, code_change/3]).
+
+-record(state, {request_queue,  %% queue()
+                worker_stack,   %% [pid()]
+                name,           %% atom()
+                module,         %% atom()
+                script_path,    %% string()
+                known_workers}). %% [pid()]
+
+%%====================================================================
+%% gen_server API
+%%====================================================================
+
+%% @spec start_link(atom(), atom(), integer(), term()) -> {ok,pid()} | ignore | {error,error()}
+%% @doc Starts the server
+%%    Name: name under which to register the manager server - this is
+%%          the Name you'll use to call {@link process/2} later
+%%    Module: module that implements the worker logic
+%%    Count: number of workers to start
+%%    Path: the command-line to run to start the worker
+start_link(Name, Module, Count, Path) when is_atom(Name),
+                                           is_atom(Module),
+                                           is_integer(Count),
+                                           is_list(Path) ->
+    gen_server:start_link({local, Name}, ?MODULE,
+                          {Name, Module, Count, Path},
+                          []).
+
+%%====================================================================
+%% script_worker API
+%%====================================================================
+
+%% @spec worker_available(atom(), pid()) -> ok
+%% @doc Register a worker process as ready for work
+%%    Name: the registered name of the manager server - same as the
+%%          first parameter to {@link start_link/4}
+%%    Pid: the process id of the worker
+worker_available(Name, Pid) when is_atom(Name), is_pid(Pid) ->
+    gen_server:cast(Name, {worker_available, Pid}).
+
+%%====================================================================
+%% client API
+%%====================================================================
+
+%% @spec process(atom(), term()) -> term()
+%% @doc Submit Data to a worker for processing.
+%%    Name: the registered name of the manager server - same as the
+%%          first parameter to {@link start_link/4}
+%%    Data: the data to give the worker
+%%  `process/2' will wait for its request to finish or timeout.  If
+%%  the request finishes successfully, the retun value is the
+%%  response to the request.  If the request times out, the response
+%%  is the timeout response from {@link gen_server:call/2}.
+process(Name, Data) ->
+    gen_server:call(Name, {process, Data}).
+
+%% @spec inc_workers(atom()) -> term()
+%% @doc Spin up a new worker for the pool.
+inc_workers(Name) ->
+    gen_server:call(Name, inc_workers).
+
+%% @spec dec_workers(atom()) -> term()
+%% @doc Remove a workers from the pool.
+dec_workers(Name) ->
+    gen_server:call(Name, dec_workers).
+
+%%====================================================================
+%% gen_server callbacks
+%%====================================================================
+
+%%--------------------------------------------------------------------
+%% Function: init(Args) -> {ok, State} |
+%%                         {ok, State, Timeout} |
+%%                         ignore               |
+%%                         {stop, Reason}
+%% Description: Initiates the server
+%%--------------------------------------------------------------------
+init({Name, Module, Count, Path}) ->
+    %% trap exits so we can start new worker processes when any of
+    %% our initial set dies
+    process_flag(trap_exit, true),
+    case [ Pid || {ok, Pid} <- [ script_worker:start_link(Name, Module, Path)
+                                 || _ <- lists:seq(1, Count) ]] of
+        [] ->
+            %% if workers failed to start, then fail to start the server
+            {stop, no_workers, Name, Module, Path};
+        Workers ->
+            %% initialize queue, but leave worker stack empty - it
+            %% will fill when each worker calls worker_avialable/0
+            {ok, #state{request_queue=queue:new(),
+                        worker_stack=[], %% workers signal when ready
+                        name=Name,
+                        module=Module,
+                        script_path=Path,
+                        known_workers=Workers}}
+    end.
+
+%%--------------------------------------------------------------------
+%% Function: %% handle_call(Request, From, State) -> {reply, Reply, State} |
+%%                                      {reply, Reply, State, Timeout} |
+%%                                      {noreply, State} |
+%%                                      {noreply, State, Timeout} |
+%%                                      {stop, Reason, Reply, State} |
+%%                                      {stop, Reason, State}
+%% Description: Handling call messages
+%%--------------------------------------------------------------------
+handle_call({process, Data}, From, State) ->
+    %% client requested a spellcheck
+    case State#state.worker_stack of
+        [Worker|Rest] ->
+            %% a worker is available - start immediately
+            script_worker:process(Worker, Data, From),
+            {noreply, State#state{worker_stack=Rest}};
+        [] ->
+            %% no worker is available - wait in the queue
+            {noreply, State#state{request_queue=queue:in({Data, From},
+                                                         State#state.request_queue)}}
+    end;
+handle_call(inc_workers, _From,
+            State=#state{name=Name, module=Module, script_path=Path}) ->
+    case script_worker:start_link(Name, Module, Path) of
+        {ok, Pid} ->
+            {reply, ok, State#state{known_workers=[Pid|State#state.known_workers]}};
+        _ ->
+            {reply, error, State}
+    end;
+handle_call(dec_workers, _From, State) ->
+    case State#state.known_workers of
+        [Head|Rest] ->
+            %% kill a worker
+            script_worker:stop(Head),
+            {reply, ok, State#state{known_workers=Rest,
+                                    worker_stack=[ Pid || Pid <- State#state.worker_stack,
+                                                          Pid /= Head]}};
+        _ ->
+            %% none to kill
+            {reply, ok, State}
+    end;
+handle_call(_Request, _From, State) ->
+    Reply = ok, %% don't die if we receive bogus calls
+    {reply, Reply, State}.
+
+%%--------------------------------------------------------------------
+%% Function: handle_cast(Msg, State) -> {noreply, State} |
+%%                                      {noreply, State, Timeout} |
+%%                                      {stop, Reason, State}
+%% Description: Handling cast messages
+%%--------------------------------------------------------------------
+handle_cast({worker_available, Pid}, State) ->
+    case lists:member(Pid, State#state.known_workers) of
+        true ->
+            %% a worker process is ready to do something
+            case queue:out(State#state.request_queue) of
+                {{value, {Data, From}}, Rest} ->
+                    %% a job is waiting - start it
+                    script_worker:process(Pid, Data, From),
+                    {noreply, State#state{request_queue=Rest}};
+                {empty, _} ->
+                    %% no job is waiting - wait for some
+                    {noreply, State#state{worker_stack=[Pid|State#state.worker_stack]}}
+            end;
+        false ->
+            %% this is not a worker we started - likely
+            %% left over from a previous manager and will pick
+            %% up that manager's EXIT after this message - ignore
+            {noreply, State}
+    end;
+handle_cast(_Msg, State) ->
+    {noreply, State}.  %% ignore bogus casts
+
+%%--------------------------------------------------------------------
+%% Function: handle_info(Info, State) -> {noreply, State} |
+%%                                       {noreply, State, Timeout} |
+%%                                       {stop, Reason, State}
+%% Description: Handling all non call/cast messages
+%%--------------------------------------------------------------------
+handle_info({'EXIT', Pid, Reason}, State) ->
+    case lists:member(Pid, State#state.known_workers) of
+        true ->
+            %% a worker died - start a new worker
+            error_logger:info_msg("~p:~p noticed ~p:~p crashed - starting new worker~n",
+                                  [State#state.name, self(), State#state.module, Pid]),
+            {ok, NewPid} = script_worker:start_link(State#state.name,
+                                                    State#state.module,
+                                                    State#state.script_path),
+            %% make sure to take the old worker's pid out of the available workers list
+            {noreply, State#state{worker_stack=[ Worker || Worker <- State#state.worker_stack,
+                                                           Pid /= Worker ],
+                                  known_workers=[NewPid|lists:delete(Pid, State#state.known_workers)]}};
+        false -> 
+            %% something else we were attached to died
+            if Reason /= normal ->
+                    %% we had better shutdown ourselves
+                    {stop, {trapped_exit, Pid}, State};
+               Reason == normal ->
+                    %% something died a natural death - ignore
+                    {noreply, State}
+            end
+    end;
+handle_info(dump_state, State) ->
+    %% debugging convenience - print State to error log
+    error_logger:info_msg("~p:~p state: ~p~n", [State#state.name, self(), State]),
+    {noreply, State};
+handle_info(_, State) ->
+    {noreply, State}. %% ignore bogus messages
+
+%%--------------------------------------------------------------------
+%% Function: terminate(Reason, State) -> void()
+%% Description: This function is called by a gen_server when it is about to
+%% terminate. It should be the opposite of Module:init/1 and do any necessary
+%% cleaning up. When it returns, the gen_server terminates with Reason.
+%% The return value is ignored.
+%%--------------------------------------------------------------------
+terminate(_Reason, _State) ->
+    ok.
+
+%%--------------------------------------------------------------------
+%% Func: code_change(OldVsn, State, Extra) -> {ok, NewState}
+%% Description: Convert process state when code is changed
+%%--------------------------------------------------------------------
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+%%--------------------------------------------------------------------
+%%% Internal functions
+%%--------------------------------------------------------------------

File apps/wiki_creole/src/script_worker.erl

+%% @author Bryan Fink <bryan@basho.com>
+%% @since 8.Apr.2009
+%% @doc Worker server for {@link script_manager} system.
+%%
+%%  `script_worker' is a generic port-handler for communicating with
+%%  os-processes that 1) can be given a batch of data to process over
+%%  stdin, and 2) respond with their results as lines printed to
+%%  stdout.
+%%
+%%  To implement your script interface, write a module with the
+%%  following four functions:
+%%
+%%  `init_trigger/0' - should return `none' if your script is ready
+%%                     to receive data as soon as it is started, or
+%%                     `process_output' if your script emits some
+%%                     data before it is ready
+%%
+%%  `handle_init/1' - only needs to be implemented if `init_trigger/0'
+%%                    returned `process_output', and should return
+%%                    `done' with the script is finally ready to
+%%                    receive a request, or `continue' if the script
+%%                    is still initing.  The parameter to `handle_init'
+%%                    is the last line of output read from the script
+%%
+%%  `process/2' - will be called when the script manager chooses this
+%%                worker to process data.  The first parameter is the
+%%                port() that the script is connected to.  The
+%%                second argument is the argument that was given to
+%%                {@link script_manager:process/2}.  `process/2' use
+%%                {@link erlang:port_command/2} to send the request
+%%                to the script.  The return value of `process/2'
+%%                is considered opaque, and will be passed verbatim
+%%                to `handle_data/2'.
+%%
+%%  `handle_data/2' - will be called whenever the script prints a line
+%%                    to stdout.  The first parameter is the current
+%%                    opaque data for the request, and the second
+%%                    parameter is the line read from the port.
+%%                    'handle_data/2' should return `{done, Response}'
+%%                    if the request has completed - `Response' will
+%%                    be returned to the caller of
+%%                    {@link script_manager:process/2}.
+%%                    `handle_data/2' should return
+%%                    `{continue, NewOpaque}' if the request is still
+%%                    processing - `NewOpaque' will be handed back
+%%                    to `handle_data/2' with the next line read.
+-module(script_worker).
+
+-behaviour(gen_server).
+
+%% behaviour
+-export([behaviour_info/1]).
+
+%% API
+-export([start_link/3]).
+-export([process/3]).
+-export([stop/1]).
+
+%% gen_server callbacks
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
+         terminate/2, code_change/3]).
+
+-record(state, {manager, module, port, state}).
+-record(processing, {when_done, partial, opaque, from}).
+
+%% behaviour
+behaviour_info(callbacks) ->
+    [{init_trigger,0}, {handle_init,1}, {process,2}, {handle_data,2}].
+
+%%====================================================================
+%% script_manager/gen_server API
+%%====================================================================
+%%--------------------------------------------------------------------
+%% Function: start_link() -> {ok,Pid} | ignore | {error,Error}
+%% Description: Starts the server
+%%--------------------------------------------------------------------
+start_link(Manager, Module, ScriptPath) ->
+    gen_server:start_link(?MODULE, {Manager, Module, ScriptPath}, []).
+
+%% @spec stop(pid()) -> ok
+%% @doc Ask the worker to stop.  The worker should continue processing
+%%      the request it was given if it has not finished it yet.
+stop(Worker) ->
+    gen_server:cast(Worker, stop).
+
+%%====================================================================
+%% client API
+%%====================================================================
+
+%% @spec(pid(), term(), tuple()) -> ok
+%% @doc submit Data to Worker for processing, and request that the
+%%      response be sent to From.  From should be the "From" handed
+%%      to {@link script_manager:handle_call/3}
+process(Worker, Data, From) when is_pid(Worker), is_tuple(From) ->
+    gen_server:cast(Worker, {process, Data, From}).
+
+%%====================================================================
+%% gen_server callbacks
+%%====================================================================
+
+%%--------------------------------------------------------------------
+%% Function: init(Args) -> {ok, State} |
+%%                         {ok, State, Timeout} |
+%%                         ignore               |
+%%                         {stop, Reason}
+%% Description: Initiates the server
+%%--------------------------------------------------------------------
+init({Manager, Module, ScriptPath}) ->
+    %% trap exits so we know if the port closes unexpectedly
+    process_flag(trap_exit, true),
+    %% open port and enter init state to handle aspell startup garbage
+    State = #state{manager=Manager,
+                   module=Module,
+                   port=open_port({spawn, ScriptPath},
+                                  [use_stdio, {line, 8192}]),
+                   state=init},
+    case Module:init_trigger() of
+        none ->
+            {ok, State, 1}; %% register with manager separately
+        process_output ->
+            {ok, State} %% register with manager when ready
+    end.
+
+%%--------------------------------------------------------------------
+%% Function: %% handle_call(Request, From, State) -> {reply, Reply, State} |
+%%                                      {reply, Reply, State, Timeout} |
+%%                                      {noreply, State} |
+%%                                      {noreply, State, Timeout} |
+%%                                      {stop, Reason, Reply, State} |
+%%                                      {stop, Reason, State}
+%% Description: Handling call messages
+%%--------------------------------------------------------------------
+handle_call(_Request, _From, State) ->
+    Reply = ok, %% everything is cast or info here, but don't die on bogus calls
+    {reply, Reply, State}.
+
+%%--------------------------------------------------------------------
+%% Function: handle_cast(Msg, State) -> {noreply, State} |
+%%                                      {noreply, State, Timeout} |
+%%                                      {stop, Reason, State}
+%% Description: Handling cast messages
+%%--------------------------------------------------------------------
+handle_cast({process, Data, From}, State=#state{state=idle, port=Port,
+                                                module=Module}) ->
+    %% handle a request to process some html
+    Opaque = Module:process(Port, Data),
+    %% wait for replies from the port
+    {noreply, State#state{state=#processing{when_done=idle,
+                                            partial=[],
+                                            opaque=Opaque,
+                                            from=From}}};
+handle_cast({process, _, From}, State=#state{module=Module}) ->
+    %% check request while in an invalid state - don't die
+    error_logger:warning_msg("~p:~p received process request while processing another query",
+                             [Module, self()]),
+    gen_server:reply(From, request_collision),
+    {noreply, State};
+handle_cast(stop, State=#state{state=idle}) ->
+    {stop, normal, State};
+handle_cast(stop, State=#state{state=P=#processing{}}) ->
+    {noreply, State#state{state=P#processing{when_done=normal}}}.
+
+%%--------------------------------------------------------------------
+%% Function: handle_info(Info, State) -> {noreply, State} |
+%%                                       {noreply, State, Timeout} |
+%%                                       {stop, Reason, State}
+%% Description: Handling all non call/cast messages
+%%--------------------------------------------------------------------
+handle_info({Port, {data, {noeol, Data}}},
+            State=#state{state=P=#processing{partial=Partial}, port=Port}) ->
+    %% queue up data until the end of the line comes
+    {noreply, State#state{state=P#processing{partial=[Data|Partial]}}};
+handle_info({Port, {data, {eol, Data}}},
+            State=#state{state=P=#processing{}, port=Port, module=Module}) ->
+    %% end of line, append to queued-up no-endline data
+    Line = lists:flatten(lists:reverse([Data|P#processing.partial])),
+    case Module:handle_data(P#processing.opaque, Line) of
+        {done, Response} ->
+            %% work is done
+            gen_server:reply(P#processing.from, Response),
+            if P#processing.when_done == idle ->
+                    %% register for more work
+                    script_manager:worker_available(State#state.manager, self()),
+                    {noreply, State#state{state=idle}};
+               true ->
+                    %% we were waiting to die - die now
+                    {stop, P#processing.when_done, State}
+            end;
+        {continue, NewOpaque} ->
+            %% work is not done
+            {noreply, State#state{state=P#processing{partial=[],
+                                                     opaque=NewOpaque}}}
+    end;
+handle_info({Port, {data, {eol, Data}}},
+            State=#state{state=init, module=Module, manager=Manager, port=Port}) ->
+    %% init data (like banners, etc. most often)
+    NewState = case Module:handle_init(Data) of
+                   done ->
+                       %% register with manager after init is done
+                       script_manager:worker_available(Manager, self()),
+                       idle;
+                   continue -> init
+               end,
+    {noreply, State#state{state=NewState}};
+handle_info({'EXIT', Port, _}, State=#state{port=Port}) ->
+    %% aspell os-process died - stop this process
+    case State#state.state of
+        #processing{from=From} ->
+            %% if we were processing something, let requester know we failed
+            %% (instead of just letting them time out)
+            gen_server:reply(From, request_failed);
+        _ -> ok
+    end,
+    {stop, port_closed, State#state{port=closed}};
+handle_info({'EXIT', Pid, _Reason}, State) ->
+    %% some other process (likely the manager) died
+    case State#state.state of
+        idle ->
+            %% we weren't doing anything, so just die
+            {stop, {trapped_exit, Pid}, State};
+        P=#processing{} ->
+            %% attempt to finish the request we were processing before dying
+            {noreply, State#state{state=P#processing{when_done={trapped_exit, Pid}}}}
+    end;
+handle_info(timeout, State=#state{manager=Manager, state=init}) ->
+    %% register with manager after init is done
+    script_manager:worker_available(Manager, self()),
+    {noreply, State#state{state=idle}};
+handle_info(_, State) ->
+    {noreply, State}. %% ignore bogus messages
+
+%%--------------------------------------------------------------------
+%% Function: terminate(Reason, State) -> void()
+%% Description: This function is called by a gen_server when it is about to
+%% terminate. It should be the opposite of Module:init/1 and do any necessary
+%% cleaning up. When it returns, the gen_server terminates with Reason.
+%% The return value is ignored.
+%%--------------------------------------------------------------------
+terminate(_Reason, #state{port=Port}) when is_port(Port) ->
+    %% shutdown the port
+    port_close(Port),
+    receive
+        {Port, closed} -> ok %% port closed successfully
+    after
+        1000 -> {error, timeout} %% port hung
+    end;
+terminate(_Reason, _State) ->
+    %% port was closed already
+    ok.
+
+%%--------------------------------------------------------------------
+%% Func: code_change(OldVsn, State, Extra) -> {ok, NewState}
+%% Description: Convert process state when code is changed
+%%--------------------------------------------------------------------
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+%%--------------------------------------------------------------------
+%%% Internal functions
+%%--------------------------------------------------------------------

File apps/wiki_creole/src/wiki_creole_app.erl

+-module(wiki_creole_app).
+
+-behaviour(application).
+
+%% Application callbacks
+-export([start/2, stop/1]).
+
+%% ===================================================================
+%% Application callbacks
+%% ===================================================================
+
+start(_StartType, _StartArgs) ->
+    wiki_creole_sup:start_link().
+
+stop(_State) ->
+    ok.

File apps/wiki_creole/src/wiki_creole_sup.erl

+
+-module(wiki_creole_sup).
+
+-behaviour(supervisor).
+
+%% API
+-export([start_link/0]).
+
+%% Supervisor callbacks
+-export([init/1]).
+
+%% Helper macro for declaring children of supervisor
+-define(CHILD(I, Type), {I, {I, start_link, []}, permanent, 5000, Type, [I]}).
+
+%% ===================================================================
+%% API functions
+%% ===================================================================
+
+start_link() ->
+    supervisor:start_link({local, ?MODULE}, ?MODULE, []).
+
+%% ===================================================================
+%% Supervisor callbacks
+%% ===================================================================
+
+init([]) ->
+    CreoleConfig = [creole, creole, 2,
+                    filename:join(code:priv_dir(wiki_creole), "creole_worker.py")],
+    Creole = {script_manager,
+              {script_manager, start_link, CreoleConfig},
+              permanent, 5000, worker, [script_manager, script_worker, creole]},
+                 
+    {ok, { {one_for_one, 5, 10}, [Creole]} }.
+

File apps/wriaki/ebin/wriaki.app

 {application, wriaki,
  [
-  {description, ""},
+  {description, "Wriaki, the Riak-based Wiki"},
   {vsn, "1"},
   {modules, [
              wriaki_app,
-             wriaki_sup
+             wriaki_sup,
+             wriaki,
+             wriaki_auth,
+             diff,
+
+             wiki_resource,
+             static_resource,
+             redirect_resource,
+             session_resource,
+             user_resource,
+             login_form_resource,
+
+             article,
+             article_history,
+             session,
+             wuser,
+
+             wrq_dtl_helper,
+             article_dtl_helper,
+             wuser_dtl_helper,
+
+             account_detail_form_dtl,
+             action_line_dtl,
+             article_diff_dtl,
+             article_dtl,
+             article_editor_dtl,
+             article_history_dtl,
+             base_dtl,
+             error_404_dtl,
+             login_form_dtl,
+             user_404_dtl,
+             user_dtl
             ]},
   {registered, []},
   {applications, [
                   kernel,
-                  stdlib
+                  stdlib,
+                  riak_erlang_client,
+                  webmachine,
+                  erlydtl,
+                  wiki_creole
                  ]},
   {mod, { wriaki_app, []}},
   {env, []}

File apps/wriaki/include/wriaki.hrl

+-define(B_ARTICLE, <<"article">>).
+-define(B_HISTORY, <<"history">>).
+-define(B_ARCHIVE, <<"archive">>).
+-define(B_USER,    <<"user">>).
+-define(B_SESSION, <<"session">>).

File apps/wriaki/priv/dispatch.conf

+%% -*- erlang -*-
+{["wiki"],              redirect_resource,   "/wiki/Welcome"}.
+{["wiki",'*'],          wiki_resource,       []}.
+{[],                    redirect_resource,   "/wiki/Welcome"}.
+
+{["user"],              login_form_resource, []}.
+{["user",name],         user_resource,       []}.
+{["user",name,session], session_resource,    []}.
+
+{["static",'*'],        static_resource,     "www"}.

File apps/wriaki/priv/www/css/wriaki.css

+span.wikiletter, span.riakletter, span.bothletter {
+    font-size: larger;
+    font-weight: bold
+}
+
+span.wikiletter {
+    color: #009;
+}
+
+span.riakletter {
+    color: #900;
+}
+
+span.bothletter {
+    color: #609;
+}
+
+#logosearch {
+    float: left;
+}
+#welcome, #login {
+    float: right;
+}
+#welcome {
+    display: none;
+}
+#content {
+    clear: both;
+}
+
+#actions {
+    padding: 5px 0;
+    background: #eef;
+}
+
+#actions a {
+    padding: 5px 1em;
+    color: #000;
+}
+
+#actions a.active {
+    font-weight: bold;
+    background: #99f;
+}
+
+.leftversion {
+    background: #ccf;
+}
+.rightversion {
+    background: #cfc;
+}
+
+div.warning {
+    width:80%;
+    margin: 5px 10%;
+    background: #ffc;
+    text-align: center;
+}

File apps/wriaki/priv/www/js/wriaki.js

+function navToSearch() {
+    var P = $('#searchtext').val();
+    window.location.href = '/'+encodeURIComponent(P);
+}
+
+function articleURL() {
+    L = window.location.href;
+    return L.slice(0, L.indexOf('?'));
+}
+
+function readCookie(name) {
+    var nameEQ = name + "=";
+    var ca = document.cookie.split(';');
+    for(var i=0;i < ca.length;i++) {
+	var c = ca[i];
+	while (c.charAt(0)==' ') c = c.substring(1,c.length);
+	if (c.indexOf(nameEQ) == 0) return c.substring(nameEQ.length,c.length);
+    }
+    return null;
+}
+
+function clearCookie(name) {
+    alert("TODO: clear "+name+" cookie");
+}
+
+$(function() {
+    /* Header search buttons */
+    $('#searchbutton').click(navToSearch);
+    $('#searchtext').keyup(function(e) {
+        if (e.keyCode == 10 || e.keyCode == 13)
+            navToSearch();
+    });
+
+    /* Article editor buttons */
+    $('#editcancel').click(function() {
+        window.location.href = articleURL();
+    });
+
+    $('#editsave').click(function() {
+        var req = {
+            url: articleURL(),
+            type: 'PUT',
+            data: {
+                text:$('#edittext').val(),
+                msg:$('#editmsg').val(),
+                vclock:$('#editvclock').val()
+            },
+            success: function() { window.location.href = req.url; }
+        };
+        $.ajax(req);
+    });
+
+    /* User settings buttons */
+    $('#settingsave').click(function() {
+        var data = {};
+
+        var p = $('input[name=password]').val();
+        if (p) data.password = p;
+        
+        var e = $('input[name=email]').val();
+        if (e) data.email = e;
+
+        var b = $('input[name=bio]').val();
+        if (b) data.name = b;
+
+        var u = $('input[name=username]');
+        if (u.length) {
+            data.username = u.val();
+            if (!data.username) {
+                alert("Please choose a username.");
+                return;
+            }
+            if (!data.password) {
+                alert("Please choose a password.");
+                return;
+            }
+        }
+
+        req = {
+            url: data.username ? '/user/'+data.username : window.location.href,
+            type: 'PUT',
+            data: data,
+            success: function() {
+                if (window.location.href.indexOf('next=')) {
+                    start = window.location.href.indexOf('next=');
+                    end = window.location.href.indexOf('&', start) ||
+                        window.location.href.length;
+                    window.location.href =
+                        decodeURIComponent(
+                            window.location.href.slice(start, end));
+                }
+            },
+            error: function(req) {
+                if (req.status == 409)
+                    $('#settingserror').text('the requested username is taken');
+                else
+                    $('#settingserror').text('an unknown error occured: '+req.responseText);
+            }
+        };
+        $.ajax(req);
+    });
+
+    $('#loginbutton').click(function() {
+        var username = $('input[name=login_username]').val();
+        var password = $('input[name=login_password]').val();
+        
+        $.ajax({
+            url:'/user/'+username,
+            type:'POST',
+            data:{'password':password},
+            success:function() { window.location.href = '/'; },
+            error:function(req) {
+                $('#loginerror').text('incorrect username/password combination');
+            }
+        });
+    });
+    
+    $('#logoutbutton').click(function() {
+        
+        $.ajax({
+            url:'/user/'+readCookie('username')+'/'+readCookie('session'),
+            type:'DELETE',
+            success:function() { window.location.href = '/'; },
+            error:function(req) {
+                if (req.status == 404) //already logged out
+                    window.location.href = '/';
+            }
+        });
+    });
+
+    if (readCookie('username') && readCookie('session')) {
+        $.ajax({
+            url:'/user/'+readCookie('username')+'/'+readCookie('session'),
+            success:function() {
+                $('#login').hide();
+                $('#welcome')
+                    .find('a').text(decodeURIComponent(readCookie('username')))
+                    .attr('href', '/user/'+readCookie('username')+'?edit')
+                    .end()
+                    .css('display', 'block');
+            },
+            error:function(req) {
+                if (req.status == 404) {
+                    clearCookie('username');
+                    clearCookie('session');
+                }
+            }
+        });
+    }
+});

File apps/wriaki/src/article.erl

+-module(article).
+
+-export([fetch/2,
+         fetch_archive/3,
+         create/5,
+         create_archive/1,
+         archive_key/1,
+         get_editor/1,
+         set_editor/2,
+         get_text/1,
+         set_text/2,
+         get_message/1,
+         set_message/2,
+         get_version/1,
+         get_timestamp/1,
+         url/1,
+         article_key_from_archive_key/1]).
+
+-include("wriaki.hrl").
+
+-define(L_EDITOR, <<"editor">>).
+
+-define(F_TEXT, <<"text">>).
+-define(F_MSG, <<"message">>).
+-define(F_VERSION, <<"version">>).
+-define(F_TS, <<"timestamp">>).
+
+fetch(Client, Key) ->
+    case rhc:get(Client, ?B_ARTICLE, Key) of
+        {ok, Object} ->
+            case rec_obj:has_siblings(Object) of
+                true ->
+                    {ok, rec_obj:get_siblings(Object)};
+                false ->
+                    {ok, [Object]}
+            end;
+        Error ->
+            Error
+    end.
+
+fetch_archive(Client, ArticleKey, Version) ->
+    rhc:get(Client, ?B_ARCHIVE, archive_key(ArticleKey, Version)).
+
+create(Key, Text, Message, Vclock, Editor)
+  when is_binary(Key), is_binary(Text), is_binary(Message),
+       is_list(Vclock), is_binary(Editor) ->
+    update_version(
+      set_text(
+        set_message(
+          set_editor(
+            rec_obj:set_vclock(
+              rec_obj:create(?B_ARTICLE, Key, {struct, []}),
+              Vclock),
+            Editor),
+          Message),
+        Text)).
+
+create_archive(Article) ->
+    set_editor(
+      rec_obj:create(?B_ARCHIVE,
+                     archive_key(Article),
+                     rec_obj:get_value(Article)),
+      get_editor(Article)).
+
+archive_key(Article) ->
+    archive_key(rec_obj:key(Article), get_version(Article)).
+archive_key(ArticleKey, Version) ->
+    iolist_to_binary([Version,<<".">>,ArticleKey]).
+
+article_key_from_archive_key(ArchiveKey) ->
+    archive_key_part(ArchiveKey, 2).
+
+article_version_from_archive_key(ArchiveKey) ->
+    archive_key_part(ArchiveKey, 1).
+
+archive_key_part(ArchiveKey, Part) ->
+    {match, [Match]} = re:run(ArchiveKey,
+                              "([^.]*)\\.(.*)",
+                              [{capture, [Part], binary}]),
+    Match.
+
+url(Article) ->
+    case rec_obj:bucket(Article) of
+        ?B_ARTICLE ->
+            ["/wiki/",mochiweb_util:unquote(rec_obj:key(Article))];
+        ?B_ARCHIVE ->
+            ["/wiki/",mochiweb_util:unquote(
+                        article_key_from_archive_key(
+                          rec_obj:key(Article)))]
+    end.
+
+get_editor(Article) ->
+    Links = rec_obj:get_links(Article),
+    [Editor] = [ E || {{_, E}, T} <- Links, T =:= ?L_EDITOR],
+    Editor.
+
+set_editor(Article, Editor) ->
+    rec_obj:add_link(
+      rec_obj:remove_links(Article, ?B_USER, ?L_EDITOR),
+      {{?B_USER, Editor}, ?L_EDITOR}).
+
+get_text(Article) ->
+    rec_obj:get_json_field(Article, ?F_TEXT).
+set_text(Article, Text) when is_binary(Text) ->
+    update_version(rec_obj:set_json_field(Article, ?F_TEXT, Text)).
+
+get_message(Article) ->
+    rec_obj:get_json_field(Article, ?F_MSG).
+set_message(Article, Message) when is_binary(Message) ->
+    update_version(rec_obj:set_json_field(Article, ?F_MSG, Message)).
+
+get_version(Article) ->
+    rec_obj:get_json_field(Article, ?F_VERSION).
+
+get_timestamp(Article) ->
+    rec_obj:get_json_field(Article, ?F_TS).
+
+update_version(Article) ->
+    {MS, S, _US} = now(),
+    TS = 1000000*MS+S,
+    rec_obj:set_json_field(
+      rec_obj:set_json_field(Article, ?F_TS, TS),
+      ?F_VERSION, list_to_binary(
+                    mochihex:to_hex(erlang:phash2({get_text(Article),
+                                                   get_message(Article),
+                                                   TS})))).

File apps/wriaki/src/article_dtl_helper.erl

+%% This file is provided to you under the Apache License,
+%% Version 2.0 (the "License"); you may not use this file
+%% except in compliance with the License.  You may obtain
+%% a copy of the License at
+
+%%   http://www.apache.org/licenses/LICENSE-2.0
+
+%% Unless required by applicable law or agreed to in writing,
+%% software distributed under the License is distributed on an
+%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+%% KIND, either express or implied.  See the License for the
+%% specific language governing permissions and limitations
+%% under the License.    
+
+%% @doc Parameterized module wrapper for Article.  Allows ErlyDTL
+%%      templates to access properties of article with dotted
+%%      notation.
+%%
+%%      Article
+%%        A riak_object from either the "article" or "archive"
+%%        bucket.
+%%
+%%      V
+%%        The article revision to choose (almost always *the*
+%%        revision of Article, but useful when Article has
+%%        multiple revisions through riak siblings).
+%% @author Bryan Fink <bryan@basho.com>
+%% @copyright 2009 Basho Technologies, Inc.  All Rights Reserved.
+-module(article_dtl_helper, [ArticleVs, V]).
+
+-export([key/0,
+         path/0,
+         encoded_vclock/0,
+         text/0,
+         html/0,
+         msg/0,
+         history/0]).
+-export([has_multiple/0, tip_versions/0, selected_version/0]).
+
+-include("wriaki.hrl").