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
|
/*
Copyright (c) 2007-2009 FastMQ Inc.
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 "../bindings/c/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;
// Positions of head and tail of the pipe (in bytes).
uint64_t head;
uint64_t tail;
uint64_t last_sent_head;
// 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 message with specified size can be written to the
// pipe. If writing the message would cause high watermark to be
// exceeded, the function returns false.
bool check_write (uint64_t size_);
// 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_);
// Flush the messages downsteam.
void flush ();
// Ask pipe to terminate.
void term ();
private:
// Command handlers.
void process_pipe_term ();
// 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;
// Positions of head and tail of the pipe (in bytes).
uint64_t head;
uint64_t tail;
// 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, false, message_pipe_granularity>
{
public:
pipe_t (object_t *reader_parent_, object_t *writer_parent_,
uint64_t hwm_, uint64_t lwm_);
~pipe_t ();
reader_t reader;
writer_t writer;
private:
pipe_t (const pipe_t&);
void operator = (const pipe_t&);
};
}
#endif
|
