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
207
208
209
210
211
212
213
214
215
216
217
218
|
/*
* Copyright (c) 2011 jMonkeyEngine
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are
* met:
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
*
* * Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
*
* * Neither the name of 'jMonkeyEngine' nor the names of its contributors
* may be used to endorse or promote products derived from this software
* without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
* TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
* PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
* CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
* EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
* PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
* PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
* LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
* NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
* SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/
package com.jme3.network.base;
import com.jme3.network.ErrorListener;
import com.jme3.network.Message;
import com.jme3.network.MessageListener;
import com.jme3.network.kernel.Connector;
import com.jme3.network.kernel.ConnectorException;
import java.nio.ByteBuffer;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* Wraps a single Connector and forwards new messages
* to the supplied message dispatcher. This is used
* by DefaultClient to manage its connector objects.
* This is only responsible for message reading and provides
* no support for buffering writes.
*
* <p>This adapter assumes a simple protocol where two
* bytes define a (short) object size with the object data
* to follow. Note: this limits the size of serialized
* objects to 32676 bytes... even though, for example,
* datagram packets can hold twice that. :P</p>
*
* @version $Revision: 8944 $
* @author Paul Speed
*/
public class ConnectorAdapter extends Thread
{
private static final int OUTBOUND_BACKLOG = 16000;
private Connector connector;
private MessageListener<Object> dispatcher;
private ErrorListener<Object> errorHandler;
private AtomicBoolean go = new AtomicBoolean(true);
private BlockingQueue<ByteBuffer> outbound;
// Writes messages out on a background thread
private WriterThread writer;
// Marks the messages as reliable or not if they came
// through this connector.
private boolean reliable;
public ConnectorAdapter( Connector connector, MessageListener<Object> dispatcher,
ErrorListener<Object> errorHandler, boolean reliable )
{
super( String.valueOf(connector) );
this.connector = connector;
this.dispatcher = dispatcher;
this.errorHandler = errorHandler;
this.reliable = reliable;
setDaemon(true);
// The backlog makes sure that the outbound channel blocks once
// a certain backlog level is reached. It is set high so that it
// is only reached in the worst cases... which are usually things like
// raw throughput tests. Technically, a saturated TCP channel could
// back up quite a bit if the buffers are full and the socket has
// stalled but 16,000 messages is still a big backlog.
outbound = new ArrayBlockingQueue<ByteBuffer>(OUTBOUND_BACKLOG);
// Note: this technically adds a potential deadlock case
// with the above code where there wasn't one before. For example,
// if a TCP outbound queue fills to capacity and a client sends
// in such a way that they block TCP message handling then if the HostedConnection
// on the server is similarly blocked then the TCP network buffers may
// all get full and no outbound messages move and we forever block
// on the queue.
// However, in practice this can't really happen... or at least it's
// the sign of other really bad things.
// First, currently the server-side outbound queues are all unbounded and
// so won't ever block the handling of messages if the outbound channel is full.
// Second, there would have to be a huge amount of data backlog for this
// to ever occur anyway.
// Third, it's a sign of a really poor architecture if 16,000 messages
// can go out in a way that blocks reads.
writer = new WriterThread();
writer.start();
}
public void close()
{
go.set(false);
// Kill the writer service
writer.shutdown();
if( connector.isConnected() )
{
// Kill the connector
connector.close();
}
}
protected void dispatch( Message m )
{
dispatcher.messageReceived( null, m );
}
public void write( ByteBuffer data )
{
try {
outbound.put( data );
} catch( InterruptedException e ) {
throw new RuntimeException( "Interrupted while waiting for queue to drain", e );
}
}
protected void handleError( Exception e )
{
if( !go.get() )
return;
errorHandler.handleError( this, e );
}
public void run()
{
MessageProtocol protocol = new MessageProtocol();
try {
while( go.get() ) {
ByteBuffer buffer = connector.read();
if( buffer == null ) {
if( go.get() ) {
throw new ConnectorException( "Connector closed." );
} else {
// Just dump out because a null buffer is expected
// from a closed/closing connector
break;
}
}
protocol.addBuffer( buffer );
Message m = null;
while( (m = protocol.getMessage()) != null ) {
m.setReliable( reliable );
dispatch( m );
}
}
} catch( Exception e ) {
handleError( e );
}
}
protected class WriterThread extends Thread
{
public WriterThread()
{
super( String.valueOf(connector) + "-writer" );
}
public void shutdown()
{
interrupt();
}
private void write( ByteBuffer data )
{
try {
connector.write(data);
} catch( Exception e ) {
handleError( e );
}
}
public void run()
{
while( go.get() ) {
try {
ByteBuffer data = outbound.take();
write(data);
} catch( InterruptedException e ) {
if( !go.get() )
return;
throw new RuntimeException( "Interrupted waiting for data", e );
}
}
}
}
}
|
