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
|
/*
Copyright (c) 2007-2010 iMatix Corporation
This file is part of 0MQ.
0MQ is free software; you can redistribute it and/or modify it under
the terms of the Lesser GNU General Public License as published by
the Free Software Foundation; either version 3 of the License, or
(at your option) any later version.
0MQ is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
Lesser GNU General Public License for more details.
You should have received a copy of the Lesser GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
package org.zmq;
public class Poller {
static {
System.loadLibrary("jzmq");
}
public static final int POLLIN = 1;
public static final int POLLOUT = 2;
public static final int POLLERR = 4;
/**
* Class constructor.
*
* @param context a 0MQ context previously created.
*/
public Poller (Context context, int size) {
this.context = context;
this.size = size;
this.next = 0;
this.socket = new Socket[size];
this.event = new short[size];
this.revent = new short[size];
for (int i = 0; i < size; ++i) {
this.event[i] = (POLLIN | POLLOUT | POLLERR);
}
}
public int register (Socket socket) {
if (next >= size)
return -1;
this.socket[next] = socket;
return next++;
}
public long getTimeout () {
return this.timeout;
}
public void setTimeout (long timeout) {
this.timeout = timeout;
}
public int getSize () {
return this.size;
}
public int getNext () {
return this.next;
}
/**
* Issue a poll call.
* @return how many objects where signalled by poll().
*/
public long poll () {
if (size <= 0 || next <= 0)
return 0;
for (int i = 0; i < next; ++i) {
revent[i] = 0;
}
return run_poll(next, socket, event, revent, timeout);
}
public boolean pollin(int index) {
return poll_mask(index, POLLIN);
}
public boolean pollout(int index) {
return poll_mask(index, POLLOUT);
}
public boolean pollerr(int index) {
return poll_mask(index, POLLERR);
}
/**
* Issue a poll call on the specified 0MQ sockets.
*
* @param socket an array of 0MQ Socket objects to poll.
* @param event an array of short values specifying what to poll for.
* @param revent an array of short values with the results.
* @param timeout the maximum timeout in microseconds.
* @return how many objects where signalled by poll().
*/
private native long run_poll(int count,
Socket[] socket,
short[] event,
short[] revent,
long timeout);
/**
* Check whether a specific mask was signalled by latest poll call.
*
* @param index the index indicating the socket.
* @param mask a combination of POLLIN, POLLOUT and POLLERR.
* @return true if specific socket was signalled as specified.
*/
private boolean poll_mask(int index, int mask) {
if (mask <= 0 || index < 0 || index >= next)
return false;
return (revent[index] & mask) > 0;
}
private Context context = null;
private long timeout = 0;
private int size = 0;
private int next = 0;
private Socket[] socket = null;
private short[] event = null;
private short[] revent = null;
}
|
