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
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
|
/*
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/>.
*/
#ifndef __ZMQ_PIPE_HPP_INCLUDED__
#define __ZMQ_PIPE_HPP_INCLUDED__
#include "../include/zmq.h"
#include "stdint.hpp"
#include "i_endpoint.hpp"
#include "yarray_item.hpp"
#include "ypipe.hpp"
#include "config.hpp"
#include "object.hpp"
namespace zmq
{
class reader_t : public object_t, public yarray_item_t
{
public:
reader_t (class object_t *parent_,
uint64_t hwm_, uint64_t lwm_);
~reader_t ();
void set_pipe (class pipe_t *pipe_);
void set_endpoint (i_endpoint *endpoint_);
// Returns true if there is at least one message to read in the pipe.
bool check_read ();
// Reads a message to the underlying pipe.
bool read (zmq_msg_t *msg_);
// Ask pipe to terminate.
void term ();
private:
// Command handlers.
void process_revive ();
void process_pipe_term_ack ();
// The underlying pipe.
class pipe_t *pipe;
// Pipe writer associated with the other side of the pipe.
class writer_t *peer;
// High and low watermarks for in-memory storage (in bytes).
uint64_t hwm;
uint64_t lwm;
// Number of messages read so far.
uint64_t msgs_read;
// Endpoint (either session or socket) the pipe is attached to.
i_endpoint *endpoint;
reader_t (const reader_t&);
void operator = (const reader_t&);
};
class writer_t : public object_t, public yarray_item_t
{
public:
writer_t (class object_t *parent_,
uint64_t hwm_, uint64_t lwm_);
~writer_t ();
void set_pipe (class pipe_t *pipe_);
void set_endpoint (i_endpoint *endpoint_);
// Checks whether a message can be written to the pipe.
// If writing the message would cause high watermark to be
// exceeded, the function returns false.
bool check_write ();
// Writes a message to the underlying pipe. Returns false if the
// message cannot be written because high watermark was reached.
bool write (zmq_msg_t *msg_);
// Remove unfinished part of a message from the pipe.
void rollback ();
// Flush the messages downsteam.
void flush ();
// Ask pipe to terminate.
void term ();
private:
void process_reader_info (uint64_t msgs_read_);
// Command handlers.
void process_pipe_term ();
// Tests whether the pipe is already full.
bool pipe_full ();
// The underlying pipe.
class pipe_t *pipe;
// Pipe reader associated with the other side of the pipe.
class reader_t *peer;
// High and low watermarks for in-memory storage (in bytes).
uint64_t hwm;
uint64_t lwm;
// Last confirmed number of messages read from the pipe.
// The actual number can be higher.
uint64_t msgs_read;
// Number of messages we have written so far.
uint64_t msgs_written;
// True iff the last attempt to write a message has failed.
bool stalled;
// Endpoint (either session or socket) the pipe is attached to.
i_endpoint *endpoint;
writer_t (const writer_t&);
void operator = (const writer_t&);
};
// Message pipe.
class pipe_t : public ypipe_t <zmq_msg_t, message_pipe_granularity>
{
public:
pipe_t (object_t *reader_parent_, object_t *writer_parent_,
uint64_t hwm_);
~pipe_t ();
reader_t reader;
writer_t writer;
private:
uint64_t compute_lwm (uint64_t hwm_);
pipe_t (const pipe_t&);
void operator = (const pipe_t&);
};
}
#endif
|
