From 00438314438f3dde00b532ac5d8d28ccc35c7096 Mon Sep 17 00:00:00 2001 From: jsdelfino Date: Wed, 17 Feb 2010 04:14:31 +0000 Subject: 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 --- sca-cpp/trunk/components/chat/Makefile.am | 17 +- sca-cpp/trunk/components/chat/chat-listener.cpp | 57 ----- sca-cpp/trunk/components/chat/chat-sender.cpp | 58 ----- sca-cpp/trunk/components/chat/chat.composite | 20 +- sca-cpp/trunk/components/chat/chatter.cpp | 162 +++++++++++++ sca-cpp/trunk/components/chat/client-test.cpp | 111 +++++++++ sca-cpp/trunk/components/chat/server-test | 39 ++++ sca-cpp/trunk/components/chat/server-test.scm | 2 +- sca-cpp/trunk/components/chat/xmpp-test.cpp | 103 +++++++++ sca-cpp/trunk/components/chat/xmpp.hpp | 292 ++++++++++++++++++++++++ 10 files changed, 732 insertions(+), 129 deletions(-) delete mode 100644 sca-cpp/trunk/components/chat/chat-listener.cpp delete mode 100644 sca-cpp/trunk/components/chat/chat-sender.cpp create mode 100644 sca-cpp/trunk/components/chat/chatter.cpp create mode 100644 sca-cpp/trunk/components/chat/client-test.cpp create mode 100755 sca-cpp/trunk/components/chat/server-test create mode 100644 sca-cpp/trunk/components/chat/xmpp-test.cpp (limited to 'sca-cpp/trunk/components/chat') 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 start(unused const list& 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& params) { - const tuscany::value func(car(params)); - if (func == "start") - return tuscany::chat::start(cdr(params)); - return tuscany::mkfailure(); -} - -} 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 post(unused const list& params) { - - //TODO post the item - - return value(true); -} - -} -} - -extern "C" { - -const tuscany::value apply(const tuscany::list& params) { - const tuscany::value func(car(params)); - if (func == "post") - return tuscany::chat::post(cdr(params)); - return tuscany::mkfailure(); -} - -} 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"> - - - sample@apache.org - + + + sca1@localhost + sca1 + - - - sample@apache.org + + + sca2@localhost + sca2 + + + @@ -41,6 +46,7 @@ + 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 post(const list& params, XMPPClient& xc) { + const value to = car(car(params)); + const value val = cadr(params); + debug(to, "chat::post::jid"); + debug(val, "chat::post::value"); + const failable r = post(to, val, xc); + if (!hasContent(r)) + return mkfailure(reason(r)); + return value(mklist(to)); +} + +/** + * A relay function that posts the XMPP messages it receives to a relay component reference. + */ +class relay { +public: + relay(const lambda&)>& rel) : rel(rel) { + } + + const failable 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("post", mklist(jid), val)); + return true; + } + +private: + const lambda&)> rel; +}; + +/** + * Subscribe and listen to an XMPP session. + */ +class subscribe { +public: + subscribe(const lambda(const value&, const value&, XMPPClient&)>& l, XMPPClient& xc) : l(l), xc(xc) { + } + + const failable operator()() const { + gc_pool pool; + debug("chat::subscribe::listen"); + const failable r = listen(l, const_cast(xc)); + debug("chat::subscribe::stopped"); + return r; + } + +private: + const lambda(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& params) const { + const tuscany::value func(car(params)); + if (func == "post") + return post(cdr(params), const_cast(xc)); + + // Stop the chatter component + if (func != "stop") + return tuscany::mkfailure(); + debug("chat::chatter::stop"); + + // Disconnect and shutdown the worker thread + disconnect(const_cast(xc)); + shutdown(const_cast(w)); + debug("chat::chatter::stopped"); + + return failable(value(lambda&)>())); + } + +private: + const XMPPClient xc; + worker w; +}; + +/** + * Start the component. + */ +const failable start(const list& 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&)>()); + const list props = hasRelay? cdr(params) : params; + const value jid = ((lambda)>)car(props))(list()); + const value pass = ((lambda)>)cadr(props))(list()); + + // Create an XMPP client session + XMPPClient xc(jid, pass, false); + const failable r = connect(xc); + if (!hasContent(r)) + return mkfailure(reason(r)); + + // Listen and relay messages in a worker thread + worker w(3); + const lambda(const value&, const value&, XMPPClient&)> rl = relay(rel); + submit >(w, lambda()>(subscribe(rl, xc))); + + // Return the chatter component lambda function + return value(lambda&)>(chatter(xc, w))); +} + +} +} + +extern "C" { + +const tuscany::value apply(const tuscany::list& params) { + const tuscany::value func(car(params)); + if (func == "start" || func == "restart") + return tuscany::chat::start(cdr(params)); + return tuscany::mkfailure(); +} + +} 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 +#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 item = list() + + (list() + "name" + string("Apple")) + + (list() + "price" + string("$2.99")); +const list entry = mklist(string("item"), string("cart-53d67a61-aa5e-4e5e-8401-39edeba8b83b"), item); + +worker w(2); +bool received; + +const failable 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 operator()() const { + const lambda(const value&, const value&, XMPPClient&)> l(listener); + listen(l, xc); + return true; + } +}; + +bool testListen() { + received = false; + XMPPClient& xc = *(new (gc_new()) XMPPClient(jid3, pass3)); + const failable c = connect(xc); + assert(hasContent(c)); + const lambda()> subs = subscribe(xc); + submit(w, subs); + return true; +} + +bool testPost() { + gc_scoped_pool pool; + http::CURLSession ch; + const failable 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 </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 +#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 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 operator()() const { + const lambda(const value&, const value&, XMPPClient&)> l(listener); + listen(l, xc); + return true; + } +}; + +bool testListen() { + received = false; + XMPPClient& xc = *(new (gc_new()) XMPPClient(jid2, pass2)); + const failable c = connect(xc); + assert(hasContent(c)); + const lambda()> subs = subscribe(xc); + submit(w, subs); + return true; +} + +bool testPost() { + XMPPClient xc(jid1, pass1); + const failable c = connect(xc); + assert(hasContent(c)); + const failable 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 + #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 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 connect(XMPPClient& xc); + friend const failable send(const char* data, const int len, XMPPClient& xc); + friend const failable send(xmpp_stanza_t* const stanza, XMPPClient& xc); + friend const failable post(const value& to, const value& val, XMPPClient& xc); + friend const failable disconnect(XMPPClient& xc); + friend const failable listen(const lambda(const value&, const value&, XMPPClient&)>& listener, XMPPClient& xc); + + const bool owner; + xmpp_ctx_t* ctx; + xmpp_conn_t* conn; + lambda(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 r = xc.listener(value(string(from)), val, xc); + if (!hasContent(r) || !content(r)) { + // Stop listening + xc.listener = lambda(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 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 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("Couldn't connect to XMPP server"); + return true; +} + +/** + * Send a buffer on an XMPP session. + */ +const failable 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("Couldn't send stanza to XMPP server"); + } + return send(data + written, len - written, xc); +} + +/** + * Send a string on an XMPP session. + */ +const failable send(const string& data, XMPPClient& xc) { + return send(c_str(data), length(data), xc); +} + +/** + * Send a stanza on an XMPP session. + */ +const failable 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("Couldn't convert stanza to text"); + const failable 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 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 r = send(stanza, xc); + xmpp_stanza_release(stanza); + if (!hasContent(r)) + return mkfailure(reason(r)); + return true; +} + +/** + * Disconnect an XMPP session. + */ +const failable disconnect(XMPPClient& xc) { + xc.disconnecting = true; + const failable r = send("", xc); + if (!hasContent(r)) + return mkfailure(reason(r)); + return true; +} + +/** + * Listen to messages received by an XMPP client. + */ +const failable listen(const lambda(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; +} + } } -- cgit v1.2.3