Mercurial > hg > blitz_condensed

comparison src/org/dancres/blitz/remote/nio/DispatcherImpl.java @ 0:3dc0c5604566

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Initial checkin of blitz 2.0 fcs - no installer yet.
author Dan Creswell <dan.creswell@gmail.com>
date Sat, 21 Mar 2009 11:00:06 +0000
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:3dc0c5604566
1 package org.dancres.blitz.remote.nio;
2
3 import org.dancres.blitz.SpaceImpl;
4
5 import java.nio.ByteBuffer;
6 import java.nio.channels.SocketChannel;
7 import java.io.IOException;
8
9 /**
10 */
11 public class DispatcherImpl implements Dispatcher {
12 private static final int GET_HEADER = 1;
13 private static final int GET_BODY = 2;
14
15 private int _state = GET_HEADER;
16
17 private int _msgSize;
18 private int _reqId;
19 private SpaceImpl _space;
20
21 /*
22 Ultimately we should maintain a pool of buffers and request one which
23 is big enough for our message body. Once we've processed, we should
24 then return the buffer to the pool.
25 */
26 private ByteBuffer _headerBuffer = ByteBuffer.allocate(8);
27 private ByteBuffer _bodyBuffer;
28
29 DispatcherImpl(SpaceImpl aSpace) {
30 _space = aSpace;
31 }
32
33 public void process(ControlBlock aBlock) throws IOException {
34 SocketChannel myChannel = aBlock.getChannel();
35
36 if (_state == GET_HEADER) {
37 if (myChannel.read(_headerBuffer) == -1)
38 throw new IOException("Socket closed");
39
40 if (! _headerBuffer.hasRemaining()) {
41 _state = GET_BODY;
42
43 _headerBuffer.flip();
44 _reqId = _headerBuffer.getInt();
45 _msgSize = _headerBuffer.getInt();
46
47 _bodyBuffer = ByteBuffer.allocate(_msgSize);
48
49 _headerBuffer.clear();
50 }
51 }
52
53 if (_state == GET_BODY) {
54 if (myChannel.read(_bodyBuffer) == -1)
55 throw new IOException("Socket closed");
56
57 if (! _bodyBuffer.hasRemaining()) {
58 // Read the message, process it
59
60 _bodyBuffer.flip();
61
62 try {
63 Pool.execute(new MessageDecoder(_reqId, _space, aBlock,
64 _bodyBuffer.array()));
65 } catch (InterruptedException anIE) {
66 System.err.println("Interrupted during message dispatch");
67 anIE.printStackTrace(System.err);
68 }
69
70 // Ready for next message
71 _state = GET_HEADER;
72
73 // _bodyBuffer.clear();
74 }
75 }
76 }
77
78 }