JMS interface

JMS interfacehttp://bill.welliver.org//space/pike/JMS interfaceIt's pretty easy to take advantage of Java products from within Pike thanks to the Java module. The interface isn't completely seamless, but it's far easier than reinventing the wheel. News 17 February 2007: The <a href="/space/pike/Fins">Fins</a> application framework now has support for asynchronous message handling using JMS. So now, the barrier to using JMS and Pike is even lower!. I was able to create a message producer and consumer with the <a href="ActiveMQ" class="wiki_link_external" >http://activemq.codehaus.org</a> JMS server; it took about an hour (assuming you understand how JMS works!), and seems to work well enough. Because JMS is a specification, you can use any JMS compliant message broker here, like TIBCO EMS, MQSeries or OpenJMS. This simple message consumer uses JNDI to get the queue connection. It simply waits on for incoming messages to be delivered, then prints them out. Now Pike can play along with all of your other messaging enabled applications! <div class="code"><pre><pre> // // simple message consumer using JNDI and JMS // int main() { object x = Java.pkg["javax/naming/InitialContext"](); object factory = x->_method("lookup", "(Ljava/lang/String;)Ljava/lang/Object;")("ConnectionFactory"); object t = x->_method("lookup", "(Ljava/lang/String;)Ljava/lang/Object;")("MyQueue"); object qc = factory->createQueueConnection(); object s = qc->createQueueSession(0, 0); object sub = s->createReceiver(t); write("getting ready to receive&#110;"); qc->start(); do{ object m = sub->receive(); if(m) { write("--> message %O to %O: %O&#110;", (string)m->getJMSMessageID(), (string)m->getJMSDestination()->toString(), (string)m->getText()); } } while(1); return 0; } </pre></pre></div> Here's the message producer: <div class="code"><pre><pre> // // simple message producer using JNDI and JMS // int main() { object x = Java.pkg["javax/naming/InitialContext"](); object factory = x->_method("lookup", "(Ljava/lang/String;)Ljava/lang/Object;")("ConnectionFactory"); object t = x->_method("lookup", "(Ljava/lang/String;)Ljava/lang/Object;")("MyQueue"); write("topic: %O&#110;", (string)t->toString()); object qc = factory->createQueueConnection(); object s = qc->createQueueSession(0,0); object sub = s->createSender(t); write("getting ready to send&#110;"); int i=0; do{ object m = s->createTextMessage(); m->setText("this is a test " <ins class="inserted"> i++ </ins>".&#110;"); sub->send(m); sleep(1); } while(1); return 0; } </pre></pre></div> For those of you who like to do things the hard way (perhaps squeezing a little extra performance in the process, here's the full-up manual version. You'll note that the low level interface to Java is a very close mapping of JNI: <div class="code"><pre><pre> #define FINDCLASS(X) (jvm->find_class(X)||(jvm->exception_describe(),jvm->exception_clear(),error("Failed to load class " X ".&#110;"),0)) static constant jvm = Java.machine; static object initial_context_class = FINDCLASS("javax/naming/InitialContext"); static object queue_connection_factory_class = FINDCLASS("javax/jms/QueueConnectionFactory"); static object queue_connection_class = FINDCLASS("javax/jms/QueueConnection"); static object queue_session_class = FINDCLASS("javax/jms/QueueSession"); static object queue_receiver_class = FINDCLASS("javax/jms/QueueReceiver"); static object message_class = FINDCLASS("javax/jms/TextMessage"); static object initial_context_init = initial_context_class->get_method("<init>", "()V"); static object initial_context_lookup = initial_context_class->get_method("lookup", "(Ljava/lang/String;)Ljava/lang/Object;"); static object qcf_create_queue_connection = queue_connection_factory_class->get_method("createQueueConnection", "()Ljavax/jms/QueueConnection;"); static object qc_create_queue_session = queue_connection_class->get_method("createQueueSession", "(ZI)Ljavax/jms/QueueSession;"); static object qs_create_receiver = queue_session_class->get_method("createReceiver", "(Ljavax/jms/Queue;)Ljavax/jms/QueueReceiver;"); static object r_receive = queue_receiver_class->get_method("receive", "()Ljavax/jms/Message;"); static object c_start = queue_connection_class->get_method("start", "()V"); static object m_get_message_id = message_class->get_method("getJMSMessageID", "()Ljava/lang/String;"); static object m_get_destination = message_class->get_method("getJMSDestination", "()Ljavax/jms/Destination;"); static object m_get_text = message_class->get_method("getText", "()Ljava/lang/String;"); /* for exception handling */ static object stringwriter_class = FINDCLASS("java/io/StringWriter"); static object printwriter_class = FINDCLASS("java/io/PrintWriter"); static object throwable_class = FINDCLASS("java/lang/Throwable"); static object stringwriter_init = stringwriter_class->get_method("<init>", "()V"); static object printwriter_init = printwriter_class->get_method("<init>", "(Ljava/io/Writer;)V"); static object printwriter_flush = printwriter_class->get_method("flush", "()V"); static object throwable_printstacktrace = throwable_class->get_method("printStackTrace", "(Ljava/io/PrintWriter;)V"); static object throwable_getmessage = throwable_class->get_method("getMessage", "()Ljava/lang/String;"); static void check_exception() { object e = jvm->exception_occurred(); if(e) { jvm->exception_clear(); object sw = stringwriter_class->alloc(); stringwriter_init(sw); object pw = printwriter_class->alloc(); printwriter_init(pw, sw); throwable_printstacktrace(e, pw); printwriter_flush(pw); array bt = backtrace(); // FIXME: KLUDGE: Sometimes the cast fails for some reason. string s = "Unknown Java exception (StringWriter failed)"; catch { s = (string)sw; }; throw(({s, bt[..sizeof(bt)-2]})); } } int main() { object initial_context = initial_context_class->alloc(); initial_context_init->call_nonvirtual(initial_context); check_exception(); object factory = initial_context_lookup->call_nonvirtual(initial_context, "ConnectionFactory"); check_exception(); object queue = initial_context_lookup->call_nonvirtual(initial_context, "MyQueue"); check_exception(); object qc = qcf_create_queue_connection->call_nonvirtual(factory); check_exception(); object qs = qc_create_queue_session->call_nonvirtual(qc, 0, 0); check_exception(); object receiver = qs_create_receiver->call_nonvirtual(qs, queue); check_exception(); write("getting ready to receive&#110;"); c_start->call_nonvirtual(qc); check_exception(); do { object m = r_receive->call_nonvirtual(receiver); object id = m_get_message_id(m); object dest = m_get_destination(m); object text = m_get_text(m); write("--> message %O to %O: %O&#110;", (string)id, (string)dest, (string)text); } while (1); return 0; } </pre></pre></div> Fins 0.9.7http://blogs.law.harvard.edu/tech/rss