hunt.http.codec.http.stream.AbstractHttp1OutputStream source code

1 module hunt.http.codec.http.stream.AbstractHttp1OutputStream;
2 
3 import hunt.http.HttpOutputStream;
4 import hunt.http.codec.http.encode.HttpGenerator;
5 
6 import hunt.http.HttpBody;
7 import hunt.http.HttpMetaData;
8 import hunt.http.HttpRequest;
9 import hunt.http.HttpResponse;
10 
11 import hunt.io.ByteBuffer;
12 import hunt.io.BufferUtils;
13 import hunt.net.Connection;
14 import hunt.Exceptions;
15 import hunt.logging;
16 
17 import std.format;
18 
19 
20 /**
21 */
22 abstract class AbstractHttp1OutputStream : HttpOutputStream {
23 
24     this(HttpMetaData metaData, bool clientMode) {
25         super(metaData, clientMode);
26     }
27 
28     override void commit() {
29         commit(cast(ByteBuffer)null);
30     }
31 
32     protected void commit(ByteBuffer data) {
33         if (committed || closed) {
34             debug warning("connection closed already, or data committed already.");
35             return;
36         }
37 
38         Connection tcpSession = getSession();
39         if(!tcpSession.isConnected()) {
40             closed = true;
41             string msg = format("connection [id=%d] closed.", tcpSession.getId());
42             debug warningf(msg);
43             return;
44         }
45         
46         version(HUNT_HTTP_DEBUG) {
47             infof("committing data: %s", data.toString());
48         }
49 
50         HttpGenerator generator = getHttpGenerator();
51         HttpGenerator.Result generatorResult;
52         ByteBuffer header = getHeaderByteBuffer();
53 
54         generatorResult = generate(metaData, header, null, data, false);
55         if (generatorResult == HttpGenerator.Result.FLUSH && 
56             generator.getState() == HttpGenerator.State.COMMITTED) {
57             if (data !is null) {
58                 // ByteBuffer[] headerAndData = [header, data];
59                 // tcpSession.encode(headerAndData);
60                 tcpSession.encode(header);
61                 tcpSession.encode(data);
62             } else {
63                 tcpSession.encode(header);
64             }
65             committed = true;
66         } else {
67             generateHttpMessageExceptionally(generatorResult, generator.getState(), 
68                 HttpGenerator.Result.FLUSH, HttpGenerator.State.COMMITTED);
69         }
70     }
71 
72     override void write(ByteBuffer data){
73         if (closed) {
74             warningf("connection closed!");
75             return;
76         }
77 
78         if (!data.hasRemaining())
79             return;
80 
81         // The browser may close the connection forcely before receiving all data of favicon.ico.
82         Connection tcpSession = getSession();
83         if(!tcpSession.isConnected()) {
84             closed = true;
85             string msg = format("connection [id=%d] closed.", tcpSession.getId());
86             debug warningf(msg);
87             return;
88         }            
89 
90         HttpGenerator generator = getHttpGenerator();
91         HttpGenerator.Result generatorResult;
92 
93         if (!committed) {
94             commit(data);
95         } else {
96             if (generator.isChunking()) {
97                 ByteBuffer chunk = BufferUtils.allocate(HttpGenerator.CHUNK_SIZE);
98 
99                 generatorResult = generate(null, null, chunk, data, false);
100                 if (generatorResult == HttpGenerator.Result.FLUSH && 
101                     generator.getState() == HttpGenerator.State.COMMITTED) {
102                     // ByteBuffer[] chunkAndData = [chunk, data];
103                     // tcpSession.encode(chunkAndData);
104                     tcpSession.encode(chunk);
105                     tcpSession.encode(data);
106                 } else {
107                     generateHttpMessageExceptionally(generatorResult, generator.getState(), 
108                         HttpGenerator.Result.FLUSH, HttpGenerator.State.COMMITTED);
109                 }
110             } else {
111                 generatorResult = generate(null, null, null, data, false);
112                 if (generatorResult == HttpGenerator.Result.FLUSH && 
113                     generator.getState() == HttpGenerator.State.COMMITTED) {
114                     tcpSession.encode(data);
115                 } else {
116                     generateHttpMessageExceptionally(generatorResult, generator.getState(), 
117                         HttpGenerator.Result.FLUSH, HttpGenerator.State.COMMITTED);
118                 }
119             }
120         }
121     }
122 
123     override void close() {
124         if (closed) {
125             version(HUNT_HTTP_DEBUG) warning("The outstream has been closed.");
126             return;
127         }
128 
129         try {
130             version(HUNT_HTTP_DEBUG) trace("The output stream for http1 is closing...");
131 
132             if(metaData.haveBody()) {
133                 version(HUNT_HTTP_DEBUG) tracef("writting body...");
134                 HttpBody httpBody = metaData.getBody();
135                 httpBody.writeTo(this);
136             }
137 
138             HttpGenerator generator = getHttpGenerator();
139             Connection tcpSession = getSession();
140             HttpGenerator.Result generatorResult;
141 
142             if (!committed) {
143                 ByteBuffer header = getHeaderByteBuffer();
144                 generatorResult = generate(metaData, header, null, null, true);
145                 if (generatorResult == HttpGenerator.Result.FLUSH && 
146                     generator.getState() == HttpGenerator.State.COMPLETING) {
147                     tcpSession.encode(header);
148                     generateLastData(generator);
149                 } else {
150                     generateHttpMessageExceptionally(generatorResult, generator.getState(), 
151                         HttpGenerator.Result.FLUSH, HttpGenerator.State.COMPLETING);
152                 }
153                 committed = true;
154             } else {
155                 if (generator.isChunking()) {
156                     version (HUNT_HTTP_DEBUG) tracef("http1 output stream is generating chunk");
157                     generatorResult = generate(null, null, null, null, true);
158                     if (generatorResult == HttpGenerator.Result.CONTINUE && 
159                         generator.getState() == HttpGenerator.State.COMPLETING) {
160                         generatorResult = generate(null, null, null, null, true);
161                         if (generatorResult == HttpGenerator.Result.NEED_CHUNK && 
162                             generator.getState() == HttpGenerator.State.COMPLETING) {
163                             generateLastChunk(generator, tcpSession);
164                         } else if (generatorResult == HttpGenerator.Result.NEED_CHUNK_TRAILER && 
165                             generator.getState() == HttpGenerator.State.COMPLETING) {
166                             generateTrailer(generator, tcpSession);
167                         }
168                     } else {
169                         generateHttpMessageExceptionally(generatorResult, generator.getState(), 
170                             HttpGenerator.Result.CONTINUE, HttpGenerator.State.COMPLETING);
171                     }
172                 } else {
173                     generatorResult = generate(null, null, null, null, true);
174                     if (generatorResult == HttpGenerator.Result.CONTINUE && 
175                         generator.getState() == HttpGenerator.State.COMPLETING) {
176                         generateLastData(generator);
177                     } else {
178                         generateHttpMessageExceptionally(generatorResult, generator.getState(), 
179                             HttpGenerator.Result.CONTINUE, HttpGenerator.State.COMPLETING);
180                     }
181                 }
182             }
183         } finally {
184             closed = true;
185             version(HUNT_HTTP_DEBUG) infof("http1 output stream closed");
186         }
187     }
188 
189     private void generateLastChunk(HttpGenerator generator, Connection tcpSession) {
190         ByteBuffer chunk = BufferUtils.allocate(HttpGenerator.CHUNK_SIZE);
191         HttpGenerator.Result generatorResult = generate(null, null, chunk, null, true);
192         if (generatorResult == HttpGenerator.Result.FLUSH && 
193             generator.getState() == HttpGenerator.State.COMPLETING) {
194             tcpSession.encode(chunk);
195             generateLastData(generator);
196         } else {
197             generateHttpMessageExceptionally(generatorResult, generator.getState(), 
198                 HttpGenerator.Result.FLUSH, HttpGenerator.State.COMPLETING);
199         }
200     }
201 
202     private void generateTrailer(HttpGenerator generator, Connection tcpSession) {
203         ByteBuffer trailer = getTrailerByteBuffer();
204         HttpGenerator.Result generatorResult = generate(null, null, trailer, null, true);
205         if (generatorResult == HttpGenerator.Result.FLUSH && 
206             generator.getState() == HttpGenerator.State.COMPLETING) {
207             tcpSession.encode(trailer);
208             generateLastData(generator);
209         } else {
210             generateHttpMessageExceptionally(generatorResult, generator.getState(), 
211                 HttpGenerator.Result.FLUSH, HttpGenerator.State.COMPLETING);
212         }
213     }
214 
215     private void generateLastData(HttpGenerator generator) {
216         HttpGenerator.Result generatorResult = generate(null, null, null, null, true);
217         if (generator.getState() == HttpGenerator.State.END) {
218             if (generatorResult == HttpGenerator.Result.DONE) {
219                 generateHttpMessageSuccessfully();
220             } else if (generatorResult == HttpGenerator.Result.SHUTDOWN_OUT) {
221                 getSession().close();
222             } else {
223                 generateHttpMessageExceptionally(generatorResult, generator.getState(), 
224                     HttpGenerator.Result.DONE, HttpGenerator.State.END);
225             }
226         } else {
227             generateHttpMessageExceptionally(generatorResult, generator.getState(), 
228                 HttpGenerator.Result.DONE, HttpGenerator.State.END);
229         }
230     }
231 
232     protected HttpGenerator.Result generate(HttpMetaData metaData, 
233         ByteBuffer header, ByteBuffer chunk, ByteBuffer content, bool last) {
234         HttpGenerator generator = getHttpGenerator();
235         if (clientMode) {
236             return generator.generateRequest(cast(HttpRequest) metaData, header, chunk, content, last);
237         } else {
238             return generator.generateResponse(cast(HttpResponse) metaData, false, header, chunk, content, last);
239         }
240     }
241 
242     abstract protected ByteBuffer getHeaderByteBuffer();
243 
244     abstract protected ByteBuffer getTrailerByteBuffer();
245 
246     abstract protected Connection getSession();
247 
248     abstract protected HttpGenerator getHttpGenerator();
249 
250     abstract protected void generateHttpMessageSuccessfully();
251 
252     abstract protected void generateHttpMessageExceptionally(HttpGenerator.Result actualResult,
253                                                              HttpGenerator.State actualState,
254                                                              HttpGenerator.Result expectedResult,
255                                                              HttpGenerator.State expectedState);
256 
257 }