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
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
|
/*
* 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.
*/
package org.apache.thrift.transport;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.SocketTimeoutException;
import org.apache.thrift.TConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* This is the most commonly used base transport. It takes an InputStream or an OutputStream or both
* and uses it/them to perform transport operations. This allows for compatibility with all the nice
* constructs Java already has to provide a variety of types of streams.
*/
public class TIOStreamTransport extends TEndpointTransport {
private static final Logger LOGGER = LoggerFactory.getLogger(TIOStreamTransport.class.getName());
/** Underlying inputStream */
protected InputStream inputStream_ = null;
/** Underlying outputStream */
protected OutputStream outputStream_ = null;
/**
* Subclasses can invoke the default constructor and then assign the input streams in the open
* method.
*/
protected TIOStreamTransport(TConfiguration config) throws TTransportException {
super(config);
}
/**
* Subclasses can invoke the default constructor and then assign the input streams in the open
* method.
*/
protected TIOStreamTransport() throws TTransportException {
super(new TConfiguration());
}
/**
* Input stream constructor, constructs an input only transport.
*
* @param config
* @param is Input stream to read from
*/
public TIOStreamTransport(TConfiguration config, InputStream is) throws TTransportException {
super(config);
inputStream_ = is;
}
/**
* Input stream constructor, constructs an input only transport.
*
* @param is Input stream to read from
*/
public TIOStreamTransport(InputStream is) throws TTransportException {
super(new TConfiguration());
inputStream_ = is;
}
/**
* Output stream constructor, constructs an output only transport.
*
* @param config
* @param os Output stream to write to
*/
public TIOStreamTransport(TConfiguration config, OutputStream os) throws TTransportException {
super(config);
outputStream_ = os;
}
/**
* Output stream constructor, constructs an output only transport.
*
* @param os Output stream to write to
*/
public TIOStreamTransport(OutputStream os) throws TTransportException {
super(new TConfiguration());
outputStream_ = os;
}
/**
* Two-way stream constructor.
*
* @param config
* @param is Input stream to read from
* @param os Output stream to read from
*/
public TIOStreamTransport(TConfiguration config, InputStream is, OutputStream os)
throws TTransportException {
super(config);
inputStream_ = is;
outputStream_ = os;
}
/**
* Two-way stream constructor.
*
* @param is Input stream to read from
* @param os Output stream to read from
*/
public TIOStreamTransport(InputStream is, OutputStream os) throws TTransportException {
super(new TConfiguration());
inputStream_ = is;
outputStream_ = os;
}
/** @return false after close is called. */
public boolean isOpen() {
return inputStream_ != null || outputStream_ != null;
}
/** The streams must already be open. This method does nothing. */
public void open() throws TTransportException {}
/** Closes both the input and output streams. */
public void close() {
try {
if (inputStream_ != null) {
try {
inputStream_.close();
} catch (IOException iox) {
LOGGER.warn("Error closing input stream.", iox);
}
}
if (outputStream_ != null) {
try {
outputStream_.close();
} catch (IOException iox) {
LOGGER.warn("Error closing output stream.", iox);
}
}
} finally {
inputStream_ = null;
outputStream_ = null;
}
}
/** Reads from the underlying input stream if not null. */
public int read(byte[] buf, int off, int len) throws TTransportException {
if (inputStream_ == null) {
throw new TTransportException(
TTransportException.NOT_OPEN, "Cannot read from null inputStream");
}
int bytesRead;
try {
bytesRead = inputStream_.read(buf, off, len);
} catch (SocketTimeoutException ste) {
throw new TTransportException(TTransportException.TIMED_OUT, ste);
} catch (IOException iox) {
throw new TTransportException(TTransportException.UNKNOWN, iox);
}
if (bytesRead < 0) {
throw new TTransportException(TTransportException.END_OF_FILE, "Socket is closed by peer.");
}
return bytesRead;
}
/** Writes to the underlying output stream if not null. */
public void write(byte[] buf, int off, int len) throws TTransportException {
if (outputStream_ == null) {
throw new TTransportException(
TTransportException.NOT_OPEN, "Cannot write to null outputStream");
}
try {
outputStream_.write(buf, off, len);
} catch (IOException iox) {
throw new TTransportException(TTransportException.UNKNOWN, iox);
}
}
/** Flushes the underlying output stream if not null. */
public void flush() throws TTransportException {
if (outputStream_ == null) {
throw new TTransportException(TTransportException.NOT_OPEN, "Cannot flush null outputStream");
}
try {
outputStream_.flush();
resetConsumedMessageSize(-1);
} catch (IOException iox) {
throw new TTransportException(TTransportException.UNKNOWN, iox);
}
}
}
|
