1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.spdy.client;
20
21 import java.io.IOException;
22 import java.nio.ByteBuffer;
23 import java.util.concurrent.Executor;
24
25 import org.eclipse.jetty.io.AbstractConnection;
26 import org.eclipse.jetty.io.ByteBufferPool;
27 import org.eclipse.jetty.io.EndPoint;
28 import org.eclipse.jetty.io.RuntimeIOException;
29 import org.eclipse.jetty.spdy.Controller;
30 import org.eclipse.jetty.spdy.ISession;
31 import org.eclipse.jetty.spdy.IdleListener;
32 import org.eclipse.jetty.spdy.StandardSession;
33 import org.eclipse.jetty.spdy.parser.Parser;
34 import org.eclipse.jetty.util.Callback;
35 import org.eclipse.jetty.util.log.Log;
36 import org.eclipse.jetty.util.log.Logger;
37
38 public class SPDYConnection extends AbstractConnection implements Controller<StandardSession.FrameBytes>, IdleListener
39 {
40 private static final Logger LOG = Log.getLogger(SPDYConnection.class);
41 private final ByteBufferPool bufferPool;
42 private final Parser parser;
43 private final int bufferSize;
44 private volatile ISession session;
45 private volatile boolean idle = false;
46
47
48 public SPDYConnection(EndPoint endPoint, ByteBufferPool bufferPool, Parser parser, Executor executor)
49 {
50 this(endPoint,bufferPool,parser,executor,8192);
51 }
52
53 public SPDYConnection(EndPoint endPoint, ByteBufferPool bufferPool, Parser parser, Executor executor,int bufferSize)
54 {
55
56 super(endPoint, executor, false);
57 this.bufferPool = bufferPool;
58 this.parser = parser;
59 onIdle(true);
60 this.bufferSize=bufferSize;
61 }
62
63 @Override
64 public void onOpen()
65 {
66 super.onOpen();
67 fillInterested();
68 }
69
70 @Override
71 public void onFillable()
72 {
73 ByteBuffer buffer = bufferPool.acquire(bufferSize, true);
74 boolean readMore = read(buffer) == 0;
75 bufferPool.release(buffer);
76 if (readMore)
77 fillInterested();
78 }
79
80 protected int read(ByteBuffer buffer)
81 {
82 EndPoint endPoint = getEndPoint();
83 while (true)
84 {
85 int filled = fill(endPoint, buffer);
86 LOG.debug("Read {} bytes", filled);
87 if (filled == 0)
88 {
89 return 0;
90 }
91 else if (filled < 0)
92 {
93 shutdown(session);
94 return -1;
95 }
96 else
97 {
98 parser.parse(buffer);
99 }
100 }
101 }
102
103 private int fill(EndPoint endPoint, ByteBuffer buffer)
104 {
105 try
106 {
107 if (endPoint.isInputShutdown())
108 return -1;
109 return endPoint.fill(buffer);
110 }
111 catch (IOException x)
112 {
113 endPoint.close();
114 throw new RuntimeIOException(x);
115 }
116 }
117
118 @Override
119 public int write(ByteBuffer buffer, final Callback<StandardSession.FrameBytes> callback, StandardSession.FrameBytes context)
120 {
121 EndPoint endPoint = getEndPoint();
122 int remaining = buffer.remaining();
123 endPoint.write(context, callback, buffer);
124 return remaining - buffer.remaining();
125 }
126
127 @Override
128 public void close()
129 {
130 goAway(session);
131 }
132
133 @Override
134 public void close(boolean onlyOutput)
135 {
136 EndPoint endPoint = getEndPoint();
137
138
139 LOG.debug("Shutting down output {}", endPoint);
140 endPoint.shutdownOutput();
141 if (!onlyOutput)
142 {
143 LOG.debug("Closing {}", endPoint);
144 endPoint.close();
145 }
146 }
147
148 @Override
149 public void onIdle(boolean idle)
150 {
151 this.idle = idle;
152 }
153
154 @Override
155 protected boolean onReadTimeout()
156 {
157 boolean idle = this.idle;
158 LOG.debug("Idle timeout on {}, idle={}", this, idle);
159 if (idle)
160 goAway(session);
161 return false;
162 }
163
164 protected void goAway(ISession session)
165 {
166 if (session != null)
167 session.goAway();
168 }
169
170 private void shutdown(ISession session)
171 {
172 if (session != null && !getEndPoint().isOutputShutdown())
173 session.shutdown();
174 }
175
176 protected ISession getSession()
177 {
178 return session;
179 }
180
181 public void setSession(ISession session)
182 {
183 this.session = session;
184 }
185 }