1 /* $Id: tstthreads.cpp,v 1.2 2005-05-20 21:32:31 adam Exp $
2 Copyright (c) 1998-2005, Index Data.
4 This file is part of the yaz-proxy.
6 YAZ proxy is free software; you can redistribute it and/or modify it under
7 the terms of the GNU General Public License as published by the Free
8 Software Foundation; either version 2, or (at your option) any later
11 YAZ proxy is distributed in the hope that it will be useful, but WITHOUT ANY
12 WARRANTY; without even the implied warranty of MERCHANTABILITY or
13 FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
16 You should have received a copy of the GNU General Public License
17 along with YAZ proxy; see the file LICENSE. If not, write to the
18 Free Software Foundation, 59 Temple Place - Suite 330, Boston, MA
31 #include <yaz++/socket-observer.h>
32 #include <yaz++/socket-manager.h>
38 virtual void destroy() = 0;
39 virtual Proxy_Msg *handle() = 0;
40 virtual void result() = 0;
43 class Proxy_Msg_Queue_List {
44 friend class Proxy_Msg_Queue;
47 Proxy_Msg_Queue_List *m_next;
50 class Proxy_Msg_Queue {
53 void enqueue(Proxy_Msg *in);
57 Proxy_Msg_Queue_List *m_list;
60 Proxy_Msg_Queue::Proxy_Msg_Queue()
65 int Proxy_Msg_Queue::size()
68 Proxy_Msg_Queue_List *l;
69 for (l = m_list; l; l = l->m_next)
74 void Proxy_Msg_Queue::enqueue(Proxy_Msg *m)
76 Proxy_Msg_Queue_List *l = new Proxy_Msg_Queue_List;
82 Proxy_Msg *Proxy_Msg_Queue::dequeue()
84 Proxy_Msg_Queue_List **l = &m_list;
89 Proxy_Msg *m = (*l)->m_item;
95 class Proxy_Thread : public IYazSocketObserver {
97 Proxy_Thread(IYazSocketObservable *obs);
98 virtual ~Proxy_Thread();
99 void socketNotify(int event);
100 void put(Proxy_Msg *m);
104 IYazSocketObservable *m_obs;
106 pthread_t m_thread_id;
107 Proxy_Msg_Queue m_input;
108 Proxy_Msg_Queue m_output;
109 pthread_mutex_t m_mutex_input_data;
110 pthread_cond_t m_cond_input_data;
111 pthread_mutex_t m_mutex_output_data;
112 pthread_cond_t m_cond_output_data;
115 static void *tfunc(void *p)
117 Proxy_Thread *pt = (Proxy_Thread *) p;
123 Proxy_Thread::Proxy_Thread(IYazSocketObservable *obs)
126 pthread_mutex_init(&m_mutex_input_data, 0);
127 pthread_cond_init(&m_cond_input_data, 0);
128 pthread_mutex_init(&m_mutex_output_data, 0);
129 pthread_cond_init(&m_cond_output_data, 0);
130 m_fd[0] = m_fd[1] = -1;
133 m_obs->addObserver(m_fd[0], this);
134 m_obs->timeoutObserver(this, 2000);
135 m_obs->maskObserver(this, YAZ_SOCKET_OBSERVE_READ);
137 pthread_create(&m_thread_id, 0, tfunc, this);
140 Proxy_Thread::~Proxy_Thread()
145 void Proxy_Thread::socketNotify(int event)
148 read(m_fd[0], buf, 1);
149 pthread_mutex_lock(&m_mutex_output_data);
150 Proxy_Msg *out = m_output.dequeue();
151 pthread_mutex_unlock(&m_mutex_output_data);
156 void Proxy_Thread::run(void *p)
160 pthread_mutex_lock(&m_mutex_input_data);
161 pthread_cond_wait(&m_cond_input_data, &m_mutex_input_data);
164 Proxy_Msg *in = m_input.dequeue();
165 pthread_mutex_unlock(&m_mutex_input_data);
168 Proxy_Msg *out = in->handle();
169 pthread_mutex_lock(&m_mutex_output_data);
170 m_output.enqueue(out);
171 pthread_cond_signal(&m_cond_output_data);
172 pthread_mutex_unlock(&m_mutex_output_data);
173 write(m_fd[1], "", 1);
175 pthread_mutex_lock(&m_mutex_input_data);
180 void Proxy_Thread::put(Proxy_Msg *m)
182 pthread_mutex_lock(&m_mutex_input_data);
184 pthread_cond_signal(&m_cond_input_data);
185 int in_size = m_input.size();
186 pthread_mutex_unlock(&m_mutex_input_data);
187 int out_size = m_output.size();
188 printf("in-size=%d out-size=%d\n", in_size, out_size);
191 class My_Msg : public Proxy_Msg {
199 class My_Thread : public Proxy_Thread {
201 My_Thread(IYazSocketObservable *obs);
204 My_Thread::My_Thread(IYazSocketObservable *obs) : Proxy_Thread(obs)
208 Proxy_Msg *My_Msg::handle()
210 My_Msg *res = new My_Msg;
214 printf("My_Msg::handle val=%d sleep=%d\n", m_val, sl);
220 void My_Msg::result()
222 printf("My_Msg::result val=%d\n", m_val);
225 void My_Msg::destroy()
230 class My_Timer_Thread : public IYazSocketObserver {
232 IYazSocketObservable *m_obs;
236 My_Timer_Thread(IYazSocketObservable *obs, My_Thread *t);
237 void socketNotify(int event);
240 My_Timer_Thread::My_Timer_Thread(IYazSocketObservable *obs,
241 My_Thread *t) : m_obs(obs)
245 obs->addObserver(m_fd[0], this);
246 obs->timeoutObserver(this, 2);
249 void My_Timer_Thread::socketNotify(int event)
252 printf("Add %d\n", seq);
253 My_Msg *m = new My_Msg;
258 int main(int argc, char **argv)
260 Yaz_SocketManager mySocketManager;
262 My_Thread m(&mySocketManager);
263 My_Timer_Thread t(&mySocketManager, &m);
264 while (mySocketManager.processEvent() > 0)