diff options
author | jsdelfino <jsdelfino@13f79535-47bb-0310-9956-ffa450edef68> | 2010-02-17 04:14:31 +0000 |
---|---|---|
committer | jsdelfino <jsdelfino@13f79535-47bb-0310-9956-ffa450edef68> | 2010-02-17 04:14:31 +0000 |
commit | 00438314438f3dde00b532ac5d8d28ccc35c7096 (patch) | |
tree | 80dbbb010c5125455a164c77670b8694231f123f /sca-cpp/trunk/components/chat | |
parent | 50063bc212e8e93d014519ef0e4d4cabef0b6be2 (diff) |
Working queue and chat components. Added a few useful start/stop scripts. Fixed lifecycle code to call start/stop/restart functions before APR pools are cleaned up in both parent and child processes. Minor build script improvements.
git-svn-id: http://svn.us.apache.org/repos/asf/tuscany@910819 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to '')
-rw-r--r-- | sca-cpp/trunk/components/chat/Makefile.am | 17 | ||||
-rw-r--r-- | sca-cpp/trunk/components/chat/chat-listener.cpp | 57 | ||||
-rw-r--r-- | sca-cpp/trunk/components/chat/chat-sender.cpp | 58 | ||||
-rw-r--r-- | sca-cpp/trunk/components/chat/chat.composite | 20 | ||||
-rw-r--r-- | sca-cpp/trunk/components/chat/chatter.cpp | 162 | ||||
-rw-r--r-- | sca-cpp/trunk/components/chat/client-test.cpp | 111 | ||||
-rwxr-xr-x | sca-cpp/trunk/components/chat/server-test | 39 | ||||
-rw-r--r-- | sca-cpp/trunk/components/chat/server-test.scm | 2 | ||||
-rw-r--r-- | sca-cpp/trunk/components/chat/xmpp-test.cpp | 103 | ||||
-rw-r--r-- | sca-cpp/trunk/components/chat/xmpp.hpp | 292 |
10 files changed, 732 insertions, 129 deletions
diff --git a/sca-cpp/trunk/components/chat/Makefile.am b/sca-cpp/trunk/components/chat/Makefile.am index ea63a475e2..11e3179e8a 100644 --- a/sca-cpp/trunk/components/chat/Makefile.am +++ b/sca-cpp/trunk/components/chat/Makefile.am @@ -17,15 +17,20 @@ if WANT_CHAT -INCLUDES = -I${LIBSTROPHE_INCLUDE} +noinst_PROGRAMS = xmpp-test client-test + +INCLUDES = -I${LIBSTROPHE_INCLUDE} -I${LIBSTROPHE_INCLUDE}/src compdir=$(prefix)/components/chat -comp_LTLIBRARIES = libchat-sender.la libchat-listener.la +comp_LTLIBRARIES = libchatter.la + +libchatter_la_SOURCES = chatter.cpp +libchatter_la_LDFLAGS = -L${LIBSTROPHE_LIB} -R${LIBSTROPHE_LIB} -lstrophe -lexpat -lssl -lresolv -libchat_sender_la_SOURCES = chat-sender.cpp -libchat_sender_la_LDFLAGS = -L${LIBSTROPHE_LIB} -R${LIBSTROPHE_LIB} -lstrophe -lexpat -lssl -lresolv +xmpp_test_SOURCES = xmpp-test.cpp +xmpp_test_LDFLAGS = -L${LIBSTROPHE_LIB} -R${LIBSTROPHE_LIB} -lstrophe -lexpat -lssl -lresolv -libchat_listener_la_SOURCES = chat-listener.cpp -libchat_listener_la_LDFLAGS = -L${LIBSTROPHE_LIB} -R${LIBSTROPHE_LIB} -lstrophe -lexpat -lssl -lresolv +client_test_SOURCES = client-test.cpp +client_test_LDFLAGS = -lxml2 -lcurl -lmozjs -L${LIBSTROPHE_LIB} -R${LIBSTROPHE_LIB} -lstrophe -lexpat -lssl -lresolv endif diff --git a/sca-cpp/trunk/components/chat/chat-listener.cpp b/sca-cpp/trunk/components/chat/chat-listener.cpp deleted file mode 100644 index 884965839a..0000000000 --- a/sca-cpp/trunk/components/chat/chat-listener.cpp +++ /dev/null @@ -1,57 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * 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. - */ - -/* $Rev$ $Date$ */ - -/** - * XMPP chat listener component implementation. - */ - -#include "string.hpp" -#include "function.hpp" -#include "list.hpp" -#include "value.hpp" -#include "monad.hpp" -#include "xmpp.hpp" - -namespace tuscany { -namespace chat { - -/** - * Initialize the component. - */ -const failable<value> start(unused const list<value>& params) { - //TODO establish a session with an XMPP server for a JID and mark the current server instance busy - - return value(true); -} - -} -} - -extern "C" { - -const tuscany::value apply(const tuscany::list<tuscany::value>& params) { - const tuscany::value func(car(params)); - if (func == "start") - return tuscany::chat::start(cdr(params)); - return tuscany::mkfailure<tuscany::value>(); -} - -} diff --git a/sca-cpp/trunk/components/chat/chat-sender.cpp b/sca-cpp/trunk/components/chat/chat-sender.cpp deleted file mode 100644 index bd67bf7315..0000000000 --- a/sca-cpp/trunk/components/chat/chat-sender.cpp +++ /dev/null @@ -1,58 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * 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. - */ - -/* $Rev$ $Date$ */ - -/** - * XMPP chat sender component implementation. - */ - -#include "string.hpp" -#include "function.hpp" -#include "list.hpp" -#include "value.hpp" -#include "monad.hpp" -#include "xmpp.hpp" - -namespace tuscany { -namespace chat { - -/** - * Post an item to an XMPP JID. - */ -const failable<value> post(unused const list<value>& params) { - - //TODO post the item - - return value(true); -} - -} -} - -extern "C" { - -const tuscany::value apply(const tuscany::list<tuscany::value>& params) { - const tuscany::value func(car(params)); - if (func == "post") - return tuscany::chat::post(cdr(params)); - return tuscany::mkfailure<tuscany::value>(); -} - -} diff --git a/sca-cpp/trunk/components/chat/chat.composite b/sca-cpp/trunk/components/chat/chat.composite index 2c19e2d0de..f02eed1418 100644 --- a/sca-cpp/trunk/components/chat/chat.composite +++ b/sca-cpp/trunk/components/chat/chat.composite @@ -22,17 +22,22 @@ targetNamespace="http://tuscany.apache.org/xmlns/sca/components" name="chat"> - <component name="chat-sender"> - <implementation.cpp path=".libs" library="libchat-sender"/> - <property name="jid">sample@apache.org</property> - <service name="chat-sender"> + <component name="print-sender"> + <implementation.cpp path=".libs" library="libchatter"/> + <property name="jid">sca1@localhost</property> + <property name="password">sca1</property> + <service name="print-sender"> <t:binding.http uri="print-sender"/> </service> </component> - <component name="chat-listener"> - <implementation.cpp path=".libs" library="libchat-listener"/> - <property name="jid">sample@apache.org</property> + <component name="print-chatter"> + <implementation.cpp path=".libs" library="libchatter"/> + <property name="jid">sca2@localhost</property> + <property name="password">sca2</property> + <service name="print-chatter"> + <t:binding.http uri="print-chatter"/> + </service> <reference name="relay" target="print"/> </component> @@ -41,6 +46,7 @@ <service name="print"> <t:binding.http uri="print"/> </service> + <reference name="report" target="print-sender"/> </component> </composite> diff --git a/sca-cpp/trunk/components/chat/chatter.cpp b/sca-cpp/trunk/components/chat/chatter.cpp new file mode 100644 index 0000000000..5a6d4909df --- /dev/null +++ b/sca-cpp/trunk/components/chat/chatter.cpp @@ -0,0 +1,162 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * 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. + */ + +/* $Rev$ $Date$ */ + +/** + * XMPP chatter component implementation. + */ + +#include "string.hpp" +#include "function.hpp" +#include "list.hpp" +#include "value.hpp" +#include "monad.hpp" +#include "parallel.hpp" +#include "xmpp.hpp" + +namespace tuscany { +namespace chat { + +/** + * Post an item to an XMPP JID. + */ +const failable<value> post(const list<value>& params, XMPPClient& xc) { + const value to = car<value>(car(params)); + const value val = cadr(params); + debug(to, "chat::post::jid"); + debug(val, "chat::post::value"); + const failable<bool> r = post(to, val, xc); + if (!hasContent(r)) + return mkfailure<value>(reason(r)); + return value(mklist<value>(to)); +} + +/** + * A relay function that posts the XMPP messages it receives to a relay component reference. + */ +class relay { +public: + relay(const lambda<value(const list<value>&)>& rel) : rel(rel) { + } + + const failable<bool> operator()(const value& jid, const value& val, unused XMPPClient& xc) const { + if (isNil(rel)) + return true; + debug(jid, "chat::relay::jid"); + debug(val, "chat::relay::value"); + const value res = rel(mklist<value>("post", mklist<value>(jid), val)); + return true; + } + +private: + const lambda<value(const list<value>&)> rel; +}; + +/** + * Subscribe and listen to an XMPP session. + */ +class subscribe { +public: + subscribe(const lambda<failable<bool>(const value&, const value&, XMPPClient&)>& l, XMPPClient& xc) : l(l), xc(xc) { + } + + const failable<bool> operator()() const { + gc_pool pool; + debug("chat::subscribe::listen"); + const failable<bool> r = listen(l, const_cast<XMPPClient&>(xc)); + debug("chat::subscribe::stopped"); + return r; + } + +private: + const lambda<failable<bool>(const value&, const value&, XMPPClient&)> l; + XMPPClient xc; +}; + +/** + * Chatter component lambda function + */ +class chatter { +public: + chatter(XMPPClient& xc, worker& w) : xc(xc), w(w) { + } + + const value operator()(const list<value>& params) const { + const tuscany::value func(car(params)); + if (func == "post") + return post(cdr(params), const_cast<XMPPClient&>(xc)); + + // Stop the chatter component + if (func != "stop") + return tuscany::mkfailure<tuscany::value>(); + debug("chat::chatter::stop"); + + // Disconnect and shutdown the worker thread + disconnect(const_cast<XMPPClient&>(xc)); + shutdown(const_cast<worker&>(w)); + debug("chat::chatter::stopped"); + + return failable<value>(value(lambda<value(const list<value>&)>())); + } + +private: + const XMPPClient xc; + worker w; +}; + +/** + * Start the component. + */ +const failable<value> start(const list<value>& params) { + // Extract the relay reference and the XMPP JID and password + const bool hasRelay = !isNil(cddr(params)); + const value rel = hasRelay? car(params) : value(lambda<value(const list<value>&)>()); + const list<value> props = hasRelay? cdr(params) : params; + const value jid = ((lambda<value(list<value>)>)car(props))(list<value>()); + const value pass = ((lambda<value(list<value>)>)cadr(props))(list<value>()); + + // Create an XMPP client session + XMPPClient xc(jid, pass, false); + const failable<bool> r = connect(xc); + if (!hasContent(r)) + return mkfailure<value>(reason(r)); + + // Listen and relay messages in a worker thread + worker w(3); + const lambda<failable<bool>(const value&, const value&, XMPPClient&)> rl = relay(rel); + submit<failable<bool> >(w, lambda<failable<bool>()>(subscribe(rl, xc))); + + // Return the chatter component lambda function + return value(lambda<value(const list<value>&)>(chatter(xc, w))); +} + +} +} + +extern "C" { + +const tuscany::value apply(const tuscany::list<tuscany::value>& params) { + const tuscany::value func(car(params)); + if (func == "start" || func == "restart") + return tuscany::chat::start(cdr(params)); + return tuscany::mkfailure<tuscany::value>(); +} + +} diff --git a/sca-cpp/trunk/components/chat/client-test.cpp b/sca-cpp/trunk/components/chat/client-test.cpp new file mode 100644 index 0000000000..f9ca2cabd6 --- /dev/null +++ b/sca-cpp/trunk/components/chat/client-test.cpp @@ -0,0 +1,111 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * 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. + */ + +/* $Rev$ $Date$ */ + +/** + * Test chat component. + */ + +#include <assert.h> +#include "stream.hpp" +#include "string.hpp" +#include "list.hpp" +#include "element.hpp" +#include "value.hpp" +#include "monad.hpp" +#include "perf.hpp" +#include "parallel.hpp" +#include "../../modules/http/curl.hpp" +#include "xmpp.hpp" + +namespace tuscany { +namespace chat { + +const value jid1("sca1@localhost"); +const value pass1("sca1"); +const value jid2("sca2@localhost"); +const value pass2("sca2"); +const value jid3("sca3@localhost"); +const value pass3("sca3"); + +const list<value> item = list<value>() + + (list<value>() + "name" + string("Apple")) + + (list<value>() + "price" + string("$2.99")); +const list<value> entry = mklist<value>(string("item"), string("cart-53d67a61-aa5e-4e5e-8401-39edeba8b83b"), item); + +worker w(2); +bool received; + +const failable<bool> listener(const value& from, const value& val, unused XMPPClient& xc) { + assert(contains(from, "sca2@localhost")); + assert(val == entry); + received = true; + return false; +} + +struct subscribe { + XMPPClient& xc; + subscribe(XMPPClient& xc) : xc(xc) { + } + const failable<bool> operator()() const { + const lambda<failable<bool>(const value&, const value&, XMPPClient&)> l(listener); + listen(l, xc); + return true; + } +}; + +bool testListen() { + received = false; + XMPPClient& xc = *(new (gc_new<XMPPClient>()) XMPPClient(jid3, pass3)); + const failable<bool> c = connect(xc); + assert(hasContent(c)); + const lambda<failable<bool>()> subs = subscribe(xc); + submit(w, subs); + return true; +} + +bool testPost() { + gc_scoped_pool pool; + http::CURLSession ch; + const failable<value> id = http::post(entry, "http://localhost:8090/print-sender/sca2@localhost", ch); + assert(hasContent(id)); + return true; +} + +bool testReceived() { + shutdown(w); + assert(received == true); + return true; +} + +} +} + +int main() { + tuscany::cout << "Testing..." << tuscany::endl; + + tuscany::chat::testListen(); + tuscany::chat::testPost(); + tuscany::chat::testReceived(); + + tuscany::cout << "OK" << tuscany::endl; + + return 0; +} diff --git a/sca-cpp/trunk/components/chat/server-test b/sca-cpp/trunk/components/chat/server-test new file mode 100755 index 0000000000..919a798fa5 --- /dev/null +++ b/sca-cpp/trunk/components/chat/server-test @@ -0,0 +1,39 @@ +#!/bin/sh + +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# 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. + +# Setup +../../modules/http/httpd-conf tmp 8090 ../../modules/http/htdocs +../../modules/server/server-conf tmp +../../modules/server/scheme-conf tmp +cat >>tmp/conf/httpd.conf <<EOF +SCAContribution `pwd`/ +SCAComposite chat.composite +EOF + +../../modules/http/httpd-start tmp +sleep 2 + +# Test +./client-test 2>/dev/null +rc=$? + +# Cleanup +../../modules/http/httpd-stop tmp +sleep 1 +return $rc diff --git a/sca-cpp/trunk/components/chat/server-test.scm b/sca-cpp/trunk/components/chat/server-test.scm index 17a42ed795..a6023708e1 100644 --- a/sca-cpp/trunk/components/chat/server-test.scm +++ b/sca-cpp/trunk/components/chat/server-test.scm @@ -17,4 +17,4 @@ ; Chat test case -(define (print x) (display x)) +(define (post key val report) (report "post" '("sca3@localhost") val)) diff --git a/sca-cpp/trunk/components/chat/xmpp-test.cpp b/sca-cpp/trunk/components/chat/xmpp-test.cpp new file mode 100644 index 0000000000..6b7fa3439f --- /dev/null +++ b/sca-cpp/trunk/components/chat/xmpp-test.cpp @@ -0,0 +1,103 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * 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. + */ + +/* $Rev$ $Date$ */ + +/** + * Test XMPP support functions. + */ + +#include <assert.h> +#include "stream.hpp" +#include "string.hpp" +#include "list.hpp" +#include "element.hpp" +#include "monad.hpp" +#include "value.hpp" +#include "perf.hpp" +#include "parallel.hpp" +#include "xmpp.hpp" + +namespace tuscany { +namespace chat { + +const value jid1("sca1@localhost"); +const value pass1("sca1"); +const value jid2("sca2@localhost"); +const value pass2("sca2"); + +worker w(2); +bool received; + +const failable<bool> listener(const value& from, const value& val, unused XMPPClient& xc) { + assert(contains(from, "sca1@localhost")); + assert(val == "hey"); + received = true; + return false; +} + +struct subscribe { + XMPPClient& xc; + subscribe(XMPPClient& xc) : xc(xc) { + } + const failable<bool> operator()() const { + const lambda<failable<bool>(const value&, const value&, XMPPClient&)> l(listener); + listen(l, xc); + return true; + } +}; + +bool testListen() { + received = false; + XMPPClient& xc = *(new (gc_new<XMPPClient>()) XMPPClient(jid2, pass2)); + const failable<bool> c = connect(xc); + assert(hasContent(c)); + const lambda<failable<bool>()> subs = subscribe(xc); + submit(w, subs); + return true; +} + +bool testPost() { + XMPPClient xc(jid1, pass1); + const failable<bool> c = connect(xc); + assert(hasContent(c)); + const failable<bool> p = post(jid2, "hey", xc); + assert(hasContent(p)); + return true; +} + +bool testReceived() { + shutdown(w); + assert(received == true); + return true; +} + +} +} + +int main() { + tuscany::cout << "Testing..." << tuscany::endl; + + tuscany::chat::testListen(); + tuscany::chat::testPost(); + tuscany::chat::testReceived(); + + tuscany::cout << "OK" << tuscany::endl; + return 0; +} diff --git a/sca-cpp/trunk/components/chat/xmpp.hpp b/sca-cpp/trunk/components/chat/xmpp.hpp index b9508f0a53..0f84d5ec4a 100644 --- a/sca-cpp/trunk/components/chat/xmpp.hpp +++ b/sca-cpp/trunk/components/chat/xmpp.hpp @@ -26,11 +26,303 @@ * XMPP support functions. */ +#include <apr_uuid.h> + #include "strophe.h" +extern "C" { +#include "common.h" +} +#include "string.hpp" +#include "list.hpp" +#include "value.hpp" +#include "monad.hpp" +#include "../../modules/scheme/eval.hpp" namespace tuscany { namespace chat { +/** + * XMPP runtime, one per process. + */ +class XMPPRuntime { +public: + XMPPRuntime() { + xmpp_initialize(); + log = xmpp_get_default_logger(XMPP_LEVEL_DEBUG); + } + + ~XMPPRuntime() { + xmpp_shutdown(); + } + +private: + friend class XMPPClient; + xmpp_log_t* log; + +} xmppRuntime; + +/** + * Represents an XMPP client. + */ +const string resourceUUID() { + apr_uuid_t uuid; + apr_uuid_get(&uuid); + char buf[APR_UUID_FORMATTED_LENGTH]; + apr_uuid_format(buf, &uuid); + return string(buf, APR_UUID_FORMATTED_LENGTH); +} + +class XMPPClient { +public: + XMPPClient(const string& jid, const string& pass, bool owner = true) : owner(owner), ctx(xmpp_ctx_new(NULL, xmppRuntime.log)), conn(xmpp_conn_new(ctx)), connecting(false), connected(false), disconnecting(false) { + xmpp_conn_set_jid(conn, c_str(jid + "/" + resourceUUID())); + xmpp_conn_set_pass(conn, c_str(pass)); + } + + XMPPClient(const XMPPClient& xc) : owner(false), ctx(xc.ctx), conn(xc.conn), listener(xc.listener), connecting(xc.connecting), connected(xc.connected), disconnecting(xc.disconnecting) { + } + + ~XMPPClient() { + extern const failable<bool> disconnect(XMPPClient& xc); + if (!owner) + return; + if (!disconnecting) + disconnect(*this); + xmpp_conn_release(conn); + xmpp_ctx_free(ctx); + } + +private: + friend int versionHandler(xmpp_conn_t* const conn, xmpp_stanza_t* const stanza, void* const udata); + friend void connHandler(xmpp_conn_t * const conn, const xmpp_conn_event_t status, const int err, xmpp_stream_error_t* const errstream, void *const udata); + friend int messageHandler(xmpp_conn_t* const conn, xmpp_stanza_t* const stanza, void* const udata); + friend const failable<bool> connect(XMPPClient& xc); + friend const failable<int> send(const char* data, const int len, XMPPClient& xc); + friend const failable<int> send(xmpp_stanza_t* const stanza, XMPPClient& xc); + friend const failable<bool> post(const value& to, const value& val, XMPPClient& xc); + friend const failable<bool> disconnect(XMPPClient& xc); + friend const failable<bool> listen(const lambda<failable<bool>(const value&, const value&, XMPPClient&)>& listener, XMPPClient& xc); + + const bool owner; + xmpp_ctx_t* ctx; + xmpp_conn_t* conn; + lambda<failable<bool>(const value&, const value&, XMPPClient&)> listener; + bool connecting; + bool connected; + bool disconnecting; +}; + +/** + * Make a text stanza. + */ +xmpp_stanza_t* textStanza(const char* text, xmpp_ctx_t* ctx) { + xmpp_stanza_t* stanza = xmpp_stanza_new(ctx); + xmpp_stanza_set_text(stanza, text); + return stanza; +} + +/** + * Make a named stanza. + */ +xmpp_stanza_t* namedStanza(const char* ns, const char* name, xmpp_ctx_t* ctx) { + xmpp_stanza_t* stanza = xmpp_stanza_new(ctx); + xmpp_stanza_set_name(stanza, name); + if (ns != NULL) + xmpp_stanza_set_ns(stanza, ns); + return stanza; +} + +/** + * Make a named stanza using a qualified name. + */ +xmpp_stanza_t* namedStanza(const char* name, xmpp_ctx_t* ctx) { + xmpp_stanza_t* stanza = xmpp_stanza_new(ctx); + xmpp_stanza_set_name(stanza, name); + return stanza; +} + +/** + * XMPP version handler. + */ +int versionHandler(xmpp_conn_t* const conn, xmpp_stanza_t* const stanza, void* const udata) { + XMPPClient& xc = *(XMPPClient*)udata; + + // Build version reply stanza + xmpp_stanza_t* reply = namedStanza("iq", xc.ctx); + xmpp_stanza_set_type(reply, "result"); + xmpp_stanza_set_id(reply, xmpp_stanza_get_id(stanza)); + xmpp_stanza_set_attribute(reply, "to", xmpp_stanza_get_attribute(stanza, "from")); + xmpp_stanza_t* query = namedStanza(xmpp_stanza_get_ns(xmpp_stanza_get_children(stanza)), "query", xc.ctx); + xmpp_stanza_add_child(reply, query); + xmpp_stanza_t* name = namedStanza("name", xc.ctx); + xmpp_stanza_add_child(query, name); + xmpp_stanza_add_child(name, textStanza("Apache Tuscany", xc.ctx)); + xmpp_stanza_t* version = namedStanza("version", xc.ctx); + xmpp_stanza_add_child(query, version); + xmpp_stanza_add_child(version, textStanza("1.0", xc.ctx)); + + // Send it + xmpp_send(conn, reply); + xmpp_stanza_release(reply); + return 1; +} + +/** + * XMPP message handler + */ +int messageHandler(unused xmpp_conn_t* const conn, xmpp_stanza_t* const stanza, void* const udata) { + // Ignore noise + if(xmpp_stanza_get_child_by_name(stanza, "body") == NULL) + return 1; + if(!strcmp(xmpp_stanza_get_attribute(stanza, "type"), "error")) + return 1; + + // Call the client listener function + XMPPClient& xc = *(XMPPClient*)udata; + const char* from = xmpp_stanza_get_attribute(stanza, "from"); + const char* text = xmpp_stanza_get_text(xmpp_stanza_get_child_by_name(stanza, "body")); + if (isNil(xc.listener)) + return 1; + const value val(scheme::readValue(text)); + debug(from, "chat::messageHandler::from"); + debug(val, "chat::messageHandler::body"); + const failable<bool> r = xc.listener(value(string(from)), val, xc); + if (!hasContent(r) || !content(r)) { + // Stop listening + xc.listener = lambda<failable<bool>(const value&, const value&, XMPPClient&)>(); + return 0; + } + return 1; +} + +/** + * XMPP connection handler. + */ +void connHandler(xmpp_conn_t * const conn, const xmpp_conn_event_t status, unused const int err, unused xmpp_stream_error_t* const errstream, void *const udata) { + XMPPClient& xc = *(XMPPClient*)udata; + xc.connecting = false; + + if (status == XMPP_CONN_CONNECT) { + debug("chat::connHandler::connected"); + xmpp_handler_add(conn, versionHandler, "jabber:iq:version", "iq", NULL, &xc); + + // Send a <presence/> stanza so that we appear online to contacts + xmpp_stanza_t* pres = xmpp_stanza_new(xc.ctx); + xmpp_stanza_set_name(pres, "presence"); + xmpp_send(conn, pres); + xmpp_stanza_release(pres); + xc.connected = true; + return; + } + + debug("chat::connHandler::disconnected"); + xc.connected = false; + if (xc.ctx->loop_status == XMPP_LOOP_RUNNING) + xc.ctx->loop_status = XMPP_LOOP_QUIT; +} + +/** + * Connect to an XMPP server. + */ +const failable<bool> connect(XMPPClient& xc) { + xc.connecting = true; + xmpp_connect_client(xc.conn, NULL, 0, connHandler, &xc); + while(xc.connecting) + xmpp_run_once(xc.ctx, 20L); + if (!xc.connected) + return mkfailure<bool>("Couldn't connect to XMPP server"); + return true; +} + +/** + * Send a buffer on an XMPP session. + */ +const failable<int> send(const char* data, const int len, XMPPClient& xc) { + if (len == 0) + return 0; + const int written = xc.conn->tls? tls_write(xc.conn->tls, data, len) : sock_write(xc.conn->sock, data, len); + if (written < 0) { + xc.conn->error = xc.conn->tls? tls_error(xc.conn->tls) : sock_error(); + return mkfailure<int>("Couldn't send stanza to XMPP server"); + } + return send(data + written, len - written, xc); +} + +/** + * Send a string on an XMPP session. + */ +const failable<int> send(const string& data, XMPPClient& xc) { + return send(c_str(data), length(data), xc); +} + +/** + * Send a stanza on an XMPP session. + */ +const failable<int> send(xmpp_stanza_t* const stanza, XMPPClient& xc) { + char *buf; + size_t len; + const int rc = xmpp_stanza_to_text(stanza, &buf, &len); + if (rc != 0) + return mkfailure<int>("Couldn't convert stanza to text"); + const failable<int> r = send(buf, len, xc); + if (!hasContent(r)) { + xmpp_free(xc.conn->ctx, buf); + return r; + } + xmpp_debug(xc.conn->ctx, "conn", "SENT: %s", buf); + xmpp_free(xc.conn->ctx, buf); + return content(r); +} + +/** + * Post a message to an XMPP jid. + */ +const failable<bool> post(const value& to, const value& val, XMPPClient& xc) { + debug(to, "chat::post::to"); + debug(val, "chat::post::body"); + + // Convert the value to a string + const string vs(scheme::writeValue(val)); + + // Build message stanza + xmpp_stanza_t* stanza = namedStanza("message", xc.ctx); + xmpp_stanza_set_type(stanza, "chat"); + xmpp_stanza_set_attribute(stanza, "to", c_str(string(to))); + xmpp_stanza_t* body = namedStanza("body", xc.ctx); + xmpp_stanza_add_child(stanza, body); + xmpp_stanza_add_child(body, textStanza(c_str(vs), xc.ctx)); + + // Send it + const failable<int> r = send(stanza, xc); + xmpp_stanza_release(stanza); + if (!hasContent(r)) + return mkfailure<bool>(reason(r)); + return true; +} + +/** + * Disconnect an XMPP session. + */ +const failable<bool> disconnect(XMPPClient& xc) { + xc.disconnecting = true; + const failable<int> r = send("</stream:stream>", xc); + if (!hasContent(r)) + return mkfailure<bool>(reason(r)); + return true; +} + +/** + * Listen to messages received by an XMPP client. + */ +const failable<bool> listen(const lambda<failable<bool>(const value&, const value&, XMPPClient&)>& listener, XMPPClient& xc) { + debug("chat::listen"); + xc.listener = listener; + xmpp_handler_add(xc.conn, messageHandler, NULL, "message", NULL, &xc); + while(xc.connected && !isNil(xc.listener)) + xmpp_run_once(xc.ctx, 1000L); + return true; +} + } } |