1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
|
#ifndef _sys_DispatchHandle_h
#define _sys_DispatchHandle_h
/*
*
* 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.
*
*/
#include "qpid/sys/Poller.h"
#include "qpid/sys/Mutex.h"
#include "qpid/CommonImportExport.h"
#include <boost/function.hpp>
#include <queue>
namespace qpid {
namespace sys {
class DispatchHandleRef;
/**
* In order to have your own handle (file descriptor on Unix) watched by the poller
* you need to:
*
* - Subclass IOHandle, in the constructor supply an appropriate
* IOHandlerPrivate object for the platform.
*
* - Construct a DispatchHandle passing it your IOHandle and
* callback functions for read, write and disconnect events.
*
* - Ensure the DispatchHandle is not deleted until the poller is no longer using it.
* TODO: astitcher document DispatchHandleRef to simplify this.
*
* When an event occurs on the handle, the poller calls the relevant callback and
* stops watching that handle. Your callback can call rewatch() or related functions
* to re-enable polling.
*/
class DispatchHandle : public PollerHandle {
friend class DispatchHandleRef;
public:
typedef boost::function1<void, DispatchHandle&> Callback;
typedef std::queue<Callback> CallbackQueue;
private:
Callback readableCallback;
Callback writableCallback;
Callback disconnectedCallback;
CallbackQueue interruptedCallbacks;
CallbackQueue callbacks; // Double buffer
Poller::shared_ptr poller;
Mutex stateLock;
enum {
IDLE,
STOPPING,
WAITING,
CALLING,
DELETING
} state;
public:
/**
* Provide a handle to poll and a set of callbacks. Note
* callbacks can be 0, meaning you are not interested in that
* event.
*
*@param h: the handle to watch. The IOHandle encapsulates a
* platfrom-specific handle to an IO object (e.g. a file descriptor
* on Unix.)
*@param rCb Callback called when the handle is readable.
*@param wCb Callback called when the handle is writable.
*@param dCb Callback called when the handle is disconnected.
*/
QPID_COMMON_EXTERN DispatchHandle(const IOHandle& h, Callback rCb, Callback wCb, Callback dCb);
QPID_COMMON_EXTERN ~DispatchHandle();
/** Add this DispatchHandle to the poller to be watched. */
QPID_COMMON_EXTERN void startWatch(Poller::shared_ptr poller);
/** Resume watching for all non-0 callbacks. */
QPID_COMMON_EXTERN void rewatch();
/** Resume watching for read only. */
QPID_COMMON_EXTERN void rewatchRead();
/** Resume watching for write only. */
QPID_COMMON_EXTERN void rewatchWrite();
/** Stop watching temporarily. The DispatchHandle remains
associated with the poller and can be re-activated using
rewatch. */
QPID_COMMON_EXTERN void unwatch();
/** Stop watching for read */
QPID_COMMON_EXTERN void unwatchRead();
/** Stop watching for write */
QPID_COMMON_EXTERN void unwatchWrite();
/** Stop watching permanently. Disassociates from the poller. */
QPID_COMMON_EXTERN void stopWatch();
/** Interrupt watching this handle and make a serialised callback that respects the
* same exclusivity guarantees as the other callbacks
*/
QPID_COMMON_EXTERN void call(Callback iCb);
protected:
QPID_COMMON_EXTERN void doDelete();
private:
QPID_COMMON_EXTERN void processEvent(Poller::EventType dir);
};
class DispatchHandleRef {
DispatchHandle* ref;
public:
typedef boost::function1<void, DispatchHandle&> Callback;
DispatchHandleRef(const IOHandle& h, Callback rCb, Callback wCb, Callback dCb) :
ref(new DispatchHandle(h, rCb, wCb, dCb))
{}
~DispatchHandleRef() { ref->doDelete(); }
void startWatch(Poller::shared_ptr poller) { ref->startWatch(poller); }
void rewatch() { ref->rewatch(); }
void rewatchRead() { ref->rewatchRead(); }
void rewatchWrite() { ref->rewatchWrite(); }
void unwatch() { ref->unwatch(); }
void unwatchRead() { ref->unwatchRead(); }
void unwatchWrite() { ref->unwatchWrite(); }
void stopWatch() { ref->stopWatch(); }
void call(Callback iCb) { ref->call(iCb); }
};
}}
#endif // _sys_DispatchHandle_h
|
