X-Git-Url: http://git.indexdata.com/?p=yazpp-moved-to-github.git;a=blobdiff_plain;f=src%2Fyaz-pdu-assoc.cpp;h=c6b166875a2b75cddc17b94d5db3886f35fe23d7;hp=744e793df97d80886bada051292059acb359079b;hb=cf75ac8890f84648d5960d4b07fea7aab9a6eff1;hpb=5c45ffbb2b430a6f41277c303a5e9b94242dab96 diff --git a/src/yaz-pdu-assoc.cpp b/src/yaz-pdu-assoc.cpp index 744e793..c6b1668 100644 --- a/src/yaz-pdu-assoc.cpp +++ b/src/yaz-pdu-assoc.cpp @@ -1,47 +1,22 @@ /* - * Copyright (c) 1998-1999, Index Data. + * Copyright (c) 1998-2001, Index Data. * See the file LICENSE for details. - * Sebastian Hammer, Adam Dickmeiss * - * $Log: yaz-pdu-assoc.cpp,v $ - * Revision 1.7 1999-04-21 12:09:01 adam - * Many improvements. Modified to proxy server to work with "sessions" - * based on cookies. - * - * Revision 1.6 1999/04/20 10:30:05 adam - * Implemented various stuff for client and proxy. Updated calls - * to ODR to reflect new name parameter. - * - * Revision 1.5 1999/04/09 11:46:57 adam - * Added object Yaz_Z_Assoc. Much more functional client. - * - * Revision 1.4 1999/03/23 14:17:57 adam - * More work on timeout handling. Work on yaz-client. - * - * Revision 1.3 1999/02/02 14:01:20 adam - * First WIN32 port of YAZ++. - * - * Revision 1.2 1999/01/28 13:08:44 adam - * Yaz_PDU_Assoc better encapsulated. Memory leak fix in - * yaz-socket-manager.cc. - * - * Revision 1.1.1.1 1999/01/28 09:41:07 adam - * First implementation of YAZ++. - * + * $Id: yaz-pdu-assoc.cpp,v 1.31 2003-10-09 12:11:10 adam Exp $ */ #include +#include +#include +#include -#include +#include -#include -#include -Yaz_PDU_Assoc::Yaz_PDU_Assoc(IYazSocketObservable *socketObservable, - COMSTACK cs) +void Yaz_PDU_Assoc::init(IYazSocketObservable *socketObservable) { m_state = Closed; - m_cs = cs; + m_cs = 0; m_socketObservable = socketObservable; m_PDU_Observer = 0; m_queue_out = 0; @@ -51,48 +26,128 @@ Yaz_PDU_Assoc::Yaz_PDU_Assoc(IYazSocketObservable *socketObservable, m_parent = 0; m_next = 0; m_destroyed = 0; + m_idleTime = 0; + m_log = LOG_DEBUG; } -IYaz_PDU_Observable *Yaz_PDU_Assoc::clone() +Yaz_PDU_Assoc::Yaz_PDU_Assoc(IYazSocketObservable *socketObservable) { - Yaz_PDU_Assoc *copy = new Yaz_PDU_Assoc(m_socketObservable, 0); - return copy; + init (socketObservable); } -Yaz_PDU_Assoc::~Yaz_PDU_Assoc() +Yaz_PDU_Assoc::Yaz_PDU_Assoc(IYazSocketObservable *socketObservable, + COMSTACK cs) { - destroy(); + init(socketObservable); + m_cs = cs; + unsigned mask = 0; + if (cs->io_pending & CS_WANT_WRITE) + mask |= YAZ_SOCKET_OBSERVE_WRITE; + if (cs->io_pending & CS_WANT_READ) + mask |= YAZ_SOCKET_OBSERVE_READ; + m_socketObservable->addObserver(cs_fileno(cs), this); + if (!mask) + { + yaz_log (m_log, "new PDU_Assoc. Ready"); + m_state = Ready; + flush_PDU(); + } + else + { + yaz_log (m_log, "new PDU_Assoc. Accepting"); + // assume comstack is accepting... + m_state = Accepting; + m_socketObservable->addObserver(cs_fileno(cs), this); + yaz_log(m_log, "maskObserver 1"); + m_socketObservable->maskObserver(this, + mask |YAZ_SOCKET_OBSERVE_EXCEPT); + } +} + + +IYaz_PDU_Observable *Yaz_PDU_Assoc::clone() +{ + Yaz_PDU_Assoc *copy = new Yaz_PDU_Assoc(m_socketObservable); + return copy; } void Yaz_PDU_Assoc::socketNotify(int event) { - logf (LOG_LOG, "Yaz_PDU_Assoc::socketNotify p=%p event = %d", this, event); - if (m_state == Connected) + yaz_log (m_log, "Yaz_PDU_Assoc::socketNotify p=%p state=%d event = %d", + this, m_state, event); + if (event & YAZ_SOCKET_OBSERVE_EXCEPT) { - m_state = Ready; - m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ| - YAZ_SOCKET_OBSERVE_EXCEPT); - m_PDU_Observer->connectNotify(); - flush_PDU(); + close(); + m_PDU_Observer->failNotify(); + return; } - else if (m_state == Connecting) + else if (event & YAZ_SOCKET_OBSERVE_TIMEOUT) { - if (event & YAZ_SOCKET_OBSERVE_READ) + m_PDU_Observer->timeoutNotify(); + return; + } + switch (m_state) + { + case Accepting: + if (!cs_accept (m_cs)) { + yaz_log (m_log, "Yaz_PDU_Assoc::cs_accept failed"); + m_cs = 0; close(); m_PDU_Observer->failNotify(); } else { - m_state = Ready; - m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ| - YAZ_SOCKET_OBSERVE_EXCEPT); - m_PDU_Observer->connectNotify(); - flush_PDU(); + unsigned mask = 0; + if (m_cs->io_pending & CS_WANT_WRITE) + mask |= YAZ_SOCKET_OBSERVE_WRITE; + if (m_cs->io_pending & CS_WANT_READ) + mask |= YAZ_SOCKET_OBSERVE_READ; + if (!mask) + { // accept is complete. turn to ready state and write if needed + m_state = Ready; + flush_PDU(); + } + else + { // accept still incomplete. + yaz_log(m_log, "maskObserver 2"); + m_socketObservable->maskObserver(this, + mask|YAZ_SOCKET_OBSERVE_EXCEPT); + } } - } - else if (m_state == Listen) - { + break; + case Connecting: + if (event & YAZ_SOCKET_OBSERVE_READ && + event & YAZ_SOCKET_OBSERVE_WRITE) + { + // For Unix: if both read and write is set, then connect failed. + close(); + m_PDU_Observer->failNotify(); + } + else + { + yaz_log (m_log, "cs_rcvconnect"); + int res = cs_rcvconnect (m_cs); + if (res == 1) + { + unsigned mask = YAZ_SOCKET_OBSERVE_EXCEPT; + if (m_cs->io_pending & CS_WANT_WRITE) + mask |= YAZ_SOCKET_OBSERVE_WRITE; + if (m_cs->io_pending & CS_WANT_READ) + mask |= YAZ_SOCKET_OBSERVE_READ; + yaz_log(m_log, "maskObserver 3"); + m_socketObservable->maskObserver(this, mask); + } + else + { + m_state = Ready; + if (m_PDU_Observer) + m_PDU_Observer->connectNotify(); + flush_PDU(); + } + } + break; + case Listen: if (event & YAZ_SOCKET_OBSERVE_READ) { int res; @@ -102,71 +157,95 @@ void Yaz_PDU_Assoc::socketNotify(int event) return; if (res < 0) { - logf(LOG_FATAL, "cs_listen failed"); + yaz_log(LOG_FATAL|LOG_ERRNO, "cs_listen failed"); return; } if (!(new_line = cs_accept(m_cs))) return; - - Yaz_PDU_Assoc *assoc = new Yaz_PDU_Assoc (m_socketObservable, - new_line); - assoc->m_parent = this; - assoc->m_next = m_children; - m_children = assoc; - - assoc->m_PDU_Observer = m_PDU_Observer->clone(assoc); - assoc->m_state = Ready; - assoc->m_socketObservable->addObserver(cs_fileno(new_line), assoc); - assoc->m_socketObservable->maskObserver(assoc, - YAZ_SOCKET_OBSERVE_READ| - YAZ_SOCKET_OBSERVE_EXCEPT); - assoc->m_socketObservable->timeoutObserver(assoc, - assoc->m_idleTime); + /* 1. create socket-manager + 2. create pdu-assoc + 3. create top-level object + setup observer for child fileid in pdu-assoc + 4. start thread + */ + yaz_log (m_log, "new session: parent fd=%d child fd=%d", + cs_fileno(m_cs), cs_fileno(new_line)); + childNotify (new_line); } - } - else if (m_state == Ready) - { - if (event & YAZ_SOCKET_OBSERVE_WRITE) - { - flush_PDU(); - } - if (event & YAZ_SOCKET_OBSERVE_READ) + break; + case Writing: + if (event & (YAZ_SOCKET_OBSERVE_READ|YAZ_SOCKET_OBSERVE_WRITE)) + flush_PDU(); + break; + case Ready: + if (event & (YAZ_SOCKET_OBSERVE_READ|YAZ_SOCKET_OBSERVE_WRITE)) { do { int res = cs_get (m_cs, &m_input_buf, &m_input_len); if (res == 1) + { + unsigned mask = YAZ_SOCKET_OBSERVE_EXCEPT; + if (m_cs->io_pending & CS_WANT_WRITE) + mask |= YAZ_SOCKET_OBSERVE_WRITE; + if (m_cs->io_pending & CS_WANT_READ) + mask |= YAZ_SOCKET_OBSERVE_READ; + yaz_log(m_log, "maskObserver 4"); + m_socketObservable->maskObserver(this, mask); return; + } else if (res <= 0) { - logf (LOG_LOG, "Connection closed by client"); + yaz_log (m_log, "Yaz_PDU_Assoc::Connection closed by peer"); close(); - m_PDU_Observer->failNotify(); + if (m_PDU_Observer) + m_PDU_Observer->failNotify(); // problem here.. return; } // lock it, so we know if recv_PDU deletes it. int destroyed = 0; m_destroyed = &destroyed; + if (!m_PDU_Observer) + return; + m_PDU_Observer->recv_PDU(m_input_buf, res); + m_destroyed = 0; if (destroyed) // it really was destroyed, return now. return; } while (m_cs && cs_more (m_cs)); + if (m_cs && m_state == Ready) + { + yaz_log(m_log, "maskObserver 5"); + m_socketObservable->maskObserver(this, + YAZ_SOCKET_OBSERVE_EXCEPT| + YAZ_SOCKET_OBSERVE_READ); + } } - if (event & YAZ_SOCKET_OBSERVE_TIMEOUT) - { - m_PDU_Observer->timeoutNotify(); - } + break; + case Closed: + yaz_log (m_log, "CLOSING state=%d event was %d", m_state, event); + close(); + m_PDU_Observer->failNotify(); + break; + default: + yaz_log (m_log, "Unknown state=%d event was %d", m_state, event); + close(); + m_PDU_Observer->failNotify(); } } void Yaz_PDU_Assoc::close() { + Yaz_PDU_Assoc *ch; + for (ch = m_children; ch; ch = ch->m_next) + ch->close(); + m_socketObservable->deleteObserver(this); m_state = Closed; if (m_cs) { - logf (LOG_LOG, "Yaz_PDU_Assoc::close fd=%d", cs_fileno(m_cs)); + yaz_log (m_log, "Yaz_PDU_Assoc::close fd=%d", cs_fileno(m_cs)); cs_close (m_cs); } m_cs = 0; @@ -176,7 +255,7 @@ void Yaz_PDU_Assoc::close() m_queue_out = m_queue_out->m_next; delete q_this; } -// free (m_input_buf); + xfree (m_input_buf); m_input_buf = 0; m_input_len = 0; } @@ -184,6 +263,7 @@ void Yaz_PDU_Assoc::close() void Yaz_PDU_Assoc::destroy() { close(); + if (m_destroyed) *m_destroyed = 1; Yaz_PDU_Assoc **c; @@ -208,11 +288,12 @@ void Yaz_PDU_Assoc::destroy() here->m_parent = 0; delete here; } + yaz_log (m_log, "Yaz_PDU_Assoc::destroy this=%p", this); } Yaz_PDU_Assoc::PDU_Queue::PDU_Queue(const char *buf, int len) { - m_buf = (char *) malloc (len); + m_buf = (char *) xmalloc (len); memcpy (m_buf, buf, len); m_len = len; m_next = 0; @@ -220,61 +301,77 @@ Yaz_PDU_Assoc::PDU_Queue::PDU_Queue(const char *buf, int len) Yaz_PDU_Assoc::PDU_Queue::~PDU_Queue() { - free (m_buf); + xfree (m_buf); } int Yaz_PDU_Assoc::flush_PDU() { int r; - logf (LOG_LOG, "Yaz_PDU_Assoc::flush_PDU"); - if (m_state != Ready) + if (m_state != Ready && m_state != Writing) { + yaz_log (m_log, "YAZ_PDU_Assoc::flush_PDU, not ready"); return 1; } PDU_Queue *q = m_queue_out; if (!q) { + m_state = Ready; + yaz_log (m_log, "YAZ_PDU_Assoc::flush_PDU queue empty"); + yaz_log(m_log, "maskObserver 6"); m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ| + YAZ_SOCKET_OBSERVE_WRITE| YAZ_SOCKET_OBSERVE_EXCEPT); return 0; } r = cs_put (m_cs, q->m_buf, q->m_len); if (r < 0) { + yaz_log (m_log, "Yaz_PDU_Assoc::flush_PDU cs_put failed"); close(); m_PDU_Observer->failNotify(); return r; } if (r == 1) { - m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ| - YAZ_SOCKET_OBSERVE_EXCEPT| - YAZ_SOCKET_OBSERVE_WRITE); - logf (LOG_LOG, "Yaz_PDU_Assoc::flush_PDU put %d bytes (incomplete)", - q->m_len); + unsigned mask = YAZ_SOCKET_OBSERVE_EXCEPT; + m_state = Writing; + if (m_cs->io_pending & CS_WANT_WRITE) + mask |= YAZ_SOCKET_OBSERVE_WRITE; + if (m_cs->io_pending & CS_WANT_READ) + mask |= YAZ_SOCKET_OBSERVE_READ; + + mask |= YAZ_SOCKET_OBSERVE_WRITE; + yaz_log(m_log, "maskObserver 7"); + m_socketObservable->maskObserver(this, mask); + yaz_log (m_log, "Yaz_PDU_Assoc::flush_PDU cs_put %d bytes fd=%d (inc)", + q->m_len, cs_fileno(m_cs)); return r; - } - logf (LOG_LOG, "Yaz_PDU_Assoc::flush_PDU put %d bytes", q->m_len); + } + yaz_log (m_log, "Yaz_PDU_Assoc::flush_PDU cs_put %d bytes", q->m_len); // whole packet sent... delete this and proceed to next ... m_queue_out = q->m_next; delete q; // don't select on write if queue is empty ... if (!m_queue_out) + { + m_state = Ready; + yaz_log(m_log, "maskObserver 8"); m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ| YAZ_SOCKET_OBSERVE_EXCEPT); + } return r; } int Yaz_PDU_Assoc::send_PDU(const char *buf, int len) { - logf (LOG_LOG, "Yaz_PDU_Assoc::send_PDU"); + yaz_log (m_log, "Yaz_PDU_Assoc::send_PDU"); PDU_Queue **pq = &m_queue_out; int is_idle = (*pq ? 0 : 1); if (!m_cs) { - logf (LOG_LOG, "Yaz_PDU_Assoc::send_PDU failed, m_cs == 0"); + yaz_log (m_log, "Yaz_PDU_Assoc::send_PDU failed, m_cs == 0"); return -1; } while (*pq) @@ -283,95 +380,105 @@ int Yaz_PDU_Assoc::send_PDU(const char *buf, int len) if (is_idle) return flush_PDU (); else - logf (LOG_LOG, "Yaz_PDU_Assoc::cannot send_PDU fd=%d", - cs_fileno(m_cs)); + yaz_log (m_log, "Yaz_PDU_Assoc::cannot send_PDU fd=%d", + cs_fileno(m_cs)); return 0; } -COMSTACK Yaz_PDU_Assoc::comstack() +COMSTACK Yaz_PDU_Assoc::comstack(const char *type_and_host, void **vp) { - if (!m_cs) - { - CS_TYPE cs_type = tcpip_type; - int protocol = PROTO_Z3950; - m_cs = cs_create (cs_type, 0, protocol); - } - return m_cs; + return cs_create_host(type_and_host, 2, vp); } void Yaz_PDU_Assoc::listen(IYaz_PDU_Observer *observer, const char *addr) { close(); - void *ap; - COMSTACK cs = comstack(); - logf (LOG_LOG, "Yaz_PDU_Assoc::listen %s", addr); + yaz_log (LOG_LOG, "Adding listener %s", addr); + m_PDU_Observer = observer; - if (!cs) - return; - ap = cs_straddr (cs, addr); - if (!ap) + void *ap; + m_cs = comstack(addr, &ap); + + if (!m_cs) return; - if (cs_bind(cs, ap, CS_SERVER) < 0) + if (cs_bind(m_cs, ap, CS_SERVER) < 0) return; - m_socketObservable->addObserver(cs_fileno(cs), this); + m_socketObservable->addObserver(cs_fileno(m_cs), this); + yaz_log(m_log, "maskObserver 9"); m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ| YAZ_SOCKET_OBSERVE_EXCEPT); + yaz_log (m_log, "Yaz_PDU_Assoc::listen ok fd=%d", cs_fileno(m_cs)); m_state = Listen; } void Yaz_PDU_Assoc::idleTime(int idleTime) { m_idleTime = idleTime; - logf (LOG_LOG, "Yaz_PDU_Assoc::idleTime(%d)", idleTime); + yaz_log (m_log, "Yaz_PDU_Assoc::idleTime(%d)", idleTime); m_socketObservable->timeoutObserver(this, m_idleTime); } void Yaz_PDU_Assoc::connect(IYaz_PDU_Observer *observer, const char *addr) { - logf (LOG_LOG, "Yaz_PDU_Assoc::connect %s", addr); + yaz_log (m_log, "Yaz_PDU_Assoc::connect %s", addr); close(); m_PDU_Observer = observer; - COMSTACK cs = comstack(); - void *ap = cs_straddr (cs, addr); - if (!ap) - { - logf (LOG_LOG, "cs_straddr failed"); - return; + void *ap; + m_cs = comstack(addr, &ap); + int res = cs_connect (m_cs, ap); + yaz_log (m_log, "Yaz_PDU_Assoc::connect fd=%d res=%d", cs_fileno(m_cs), + res); + m_socketObservable->addObserver(cs_fileno(m_cs), this); + + if (res == 0) + { // Connect complete + m_state = Connecting; + unsigned mask = YAZ_SOCKET_OBSERVE_EXCEPT; + mask |= YAZ_SOCKET_OBSERVE_WRITE; + mask |= YAZ_SOCKET_OBSERVE_READ; + yaz_log(m_log, "maskObserver 11"); + m_socketObservable->maskObserver(this, mask); } - int res = cs_connect (cs, ap); - if (res < 0) - { - logf (LOG_LOG, "Yaz_PDU_Assoc::connect failed"); -#if 1 - logf (LOG_LOG, "Yaz_PDU_Assoc::connect fd=%d", cs_fileno(cs)); - m_socketObservable->addObserver(cs_fileno(cs), this); - m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ| - YAZ_SOCKET_OBSERVE_EXCEPT| - YAZ_SOCKET_OBSERVE_WRITE); + else if (res > 0) + { // Connect pending m_state = Connecting; -#else - close (); -#endif + unsigned mask = YAZ_SOCKET_OBSERVE_EXCEPT; + if (m_cs->io_pending & CS_WANT_WRITE) + mask |= YAZ_SOCKET_OBSERVE_WRITE; + if (m_cs->io_pending & CS_WANT_READ) + mask |= YAZ_SOCKET_OBSERVE_READ; + yaz_log(m_log, "maskObserver 11"); + m_socketObservable->maskObserver(this, mask); } else - { - logf (LOG_LOG, "Yaz_PDU_Assoc::connect fd=%d", cs_fileno(cs)); - m_socketObservable->addObserver(cs_fileno(cs), this); - m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ| - YAZ_SOCKET_OBSERVE_EXCEPT| - YAZ_SOCKET_OBSERVE_WRITE); - if (res == 1) - { - logf (LOG_LOG, "Yaz_PDU_Assoc::connect pending"); - m_state = Connecting; - } - else - { - logf (LOG_LOG, "Yaz_PDU_Assoc::Connect complete"); - m_state = Connected; - } + { // Connect failed immediately + // Since m_state is Closed we can distinguish this case from + // normal connect in socketNotify handler + yaz_log(m_log, "maskObserver 12"); + m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_WRITE| + YAZ_SOCKET_OBSERVE_EXCEPT); } } + +// Single-threaded... Only useful for non-blocking handlers +void Yaz_PDU_Assoc::childNotify(COMSTACK cs) +{ + Yaz_PDU_Assoc *new_observable = + new Yaz_PDU_Assoc (m_socketObservable, cs); + + new_observable->m_next = m_children; + m_children = new_observable; + new_observable->m_parent = this; + + // Clone PDU Observer + new_observable->m_PDU_Observer = m_PDU_Observer->sessionNotify + (new_observable, cs_fileno(cs)); +} + +const char*Yaz_PDU_Assoc::getpeername() +{ + return cs_addrstr(m_cs); +}