java/org/apache/tomcat/spdy/SpdyStream.java - tomcat80 - Git at Google

 /*
  *  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.tomcat.spdy;

 import java.io.IOException;
 import java.io.InputStream;
 import java.io.PrintWriter;
 import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;

 /**
  * One SPDY stream.
  *
  * Created by SpdyContext.getProcessor(framer).
  *
  * The methods are called in a IO thread when the framer received a frame for
  * this stream.
  *
  * They should not block.
  *
  * The frame must be either consumed or popInFrame must be called, after the
  * call is done the frame will be reused.
  */
 public class SpdyStream implements Runnable {
     private final SpdyConnection spdy;

     public SpdyFrame reqFrame;

     public SpdyFrame resFrame;

     /**
      * For blocking support.
      */
     private final BlockingQueue<SpdyFrame> inData = new LinkedBlockingQueue<>();

     protected boolean finSent;

     protected boolean finRcvd;

     /**
      *  Dummy data frame to insert on reset / go away
      */
     private static final SpdyFrame END_FRAME;

     static {
         END_FRAME = new SpdyFrame(16);
         END_FRAME.endData = 0;
         END_FRAME.off = 0;
         END_FRAME.c = false;
         END_FRAME.flags =SpdyConnection.FLAG_HALF_CLOSE;
     }

     public SpdyStream(SpdyConnection spdy) {
         this.spdy = spdy;
     }

     public void dump(PrintWriter out) {
         if (reqFrame != null) {
             out.println("Req: " + reqFrame);
         }
         if (resFrame != null) {
             out.println("Res: " + resFrame);
         }
         out.println("In: " + inData.size() + (finRcvd ? " FIN":""));
     }

     /**
      * Non-blocking, called when a data frame is received.
      */
     public void onDataFrame(SpdyFrame inFrame) {
         synchronized(this) {
             inData.add(inFrame);
             if (inFrame.closed()) {
                 finRcvd = true;
             }
         }
     }

     /**
      * Non-blocking - handles a syn stream package. The processor must consume
      * frame.data or set it to null.
      *
      * The base method is for client implementation - servers need to override
      * and process the frame as a request.
      */
     public void onCtlFrame(SpdyFrame frame) {
         // TODO: handle RST
         if (frame.type == SpdyConnection.TYPE_SYN_STREAM) {
             reqFrame = frame;
         } else if (frame.type == SpdyConnection.TYPE_SYN_REPLY) {
             resFrame = frame;
         }
         synchronized (this) {
             inData.add(frame);
             if (frame.isHalfClose()) {
                 finRcvd = true;
             }
         }
     }

     /**
      * Called on GOAWAY or reset.
      */
     public void onReset() {
         finRcvd = true;
         finSent = true;

         // To unblock
         inData.add(END_FRAME);
     }

     /**
      * True if the channel both received and sent FIN frames.
      *
      * This is tracked by the processor, to avoid extra storage in framer.
      */
     public boolean isFinished() {
         return finSent && finRcvd;
     }

     /**
      * Waits and return the next data frame, null on timeout.
      */
     public SpdyFrame getDataFrame(long to) throws IOException {
         while (true) {
             SpdyFrame res = getFrame(to);
             if (res == null || res.isData()) {
                 return res;
             }
             if (res.type == SpdyConnection.TYPE_RST_STREAM) {
                 throw new IOException("Reset");
             }
         }
     }

     /**
      * Waits and return the next frame.
      *
      * First frame will be the control frame
      */
     public SpdyFrame getFrame(long to) {
         SpdyFrame in;
         try {
             synchronized (this) {
                 if (inData.size() == 0 && finRcvd) {
                     return END_FRAME;
                 }
             }
             in = inData.poll(to, TimeUnit.MILLISECONDS);
             return in;
         } catch (InterruptedException e) {
             return null;
         }
     }

     public void getHeaders(Map<String, String> resHeaders) {
         SpdyFrame f = resFrame;
         int nvCount = f.nvCount;
         for (int i = 0; i < nvCount; i++) {
             int len = f.read16();
             String n = new String(f.data, f.off, len, StandardCharsets.UTF_8);
             f.advance(len);
             len = f.read16();
             String v = new String(f.data, f.off, len, StandardCharsets.UTF_8);
             f.advance(len);
             resHeaders.put(n, v);
         }
     }

     public SpdyFrame getRequest() {
         if (reqFrame == null) {
             reqFrame = spdy.getFrame(SpdyConnection.TYPE_SYN_STREAM);
         }
         return reqFrame;
     }

     public SpdyFrame getResponse() {
         if (resFrame == null) {
             resFrame = spdy.getFrame(SpdyConnection.TYPE_SYN_REPLY);
             resFrame.streamId = reqFrame.streamId;
         }
         return resFrame;
     }

     public synchronized void sendDataFrame(byte[] data, int start,
             int length, boolean close) {

         SpdyFrame oframe = spdy.getDataFrame();

         // Options:
         // 1. wrap the byte[] data, use a separate header[], wait frame sent
         // -> 2 socket writes
         // 2. copy the data to frame byte[] -> non-blocking queue
         // 3. copy the data, blocking drain -> like 1, trade one copy to
         // avoid
         // 1 tcp packet. That's the current choice, seems closer to rest of
         // tomcat

         if (close)
             oframe.halfClose();

         oframe.append(data, start, length);
         spdy.send(oframe, this);
     }

     public void send() {
         send("http", "GET");
     }

     public void send(String host, String url, String scheme, String method) {
         getRequest().addHeader("host", host);
         getRequest().addHeader("url", url);

         send(scheme, method);
     }

     public void send(String scheme, String method) {
         getRequest();
         if ("GET".equalsIgnoreCase(method)) {
             // TODO: add the others
             reqFrame.halfClose();
         }
         getRequest().addHeader("scheme", scheme);
         getRequest().addHeader("method", method);
         getRequest().addHeader("version", "HTTP/1.1");
         if (reqFrame.isHalfClose()) {
             finSent = true;
         }
         spdy.send(reqFrame, this);
     }

     @Override
     public void run() {
         try {
             spdy.spdyContext.handler.onStream(spdy, this);
         } catch (IOException e) {
             e.printStackTrace();
             // TODO: send rst, error processing the stream.
         }
     }


     public InputStream getInputStream() {
         return new SpdyInputStream();
     }

     private class SpdyInputStream extends InputStream {
         private SpdyFrame current = null;
         private long to = 10000; // TODO

         private void fill() {
             if (current == null || current.off == current.endData) {
                 current = getFrame(to);
             }
         }

         @Override
         public int read() throws IOException {
             fill();
             if (current == null) {
                 return -1;
             }
             return current.readByte();
         }

         @Override
         public int read(byte b[], int off, int len) throws IOException {
             fill();
             if (current == null) {
                 return -1;
             }
             // don't wait for next frame
             int rd = Math.min(len, current.endData - current.off);
             System.arraycopy(current.data, current.off, b, off, rd);
             current.off += rd;
             return rd;
         }

         @Override
         public int available() throws IOException {
             return 0;
         }
         @Override
         public void close() throws IOException {
             // send RST if not closed
         }


     }
 }
	/*
	* 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.tomcat.spdy;

	import java.io.IOException;
	import java.io.InputStream;
	import java.io.PrintWriter;
	import java.nio.charset.StandardCharsets;
	import java.util.Map;
	import java.util.concurrent.BlockingQueue;
	import java.util.concurrent.LinkedBlockingQueue;
	import java.util.concurrent.TimeUnit;

	/**
	* One SPDY stream.
	*
	* Created by SpdyContext.getProcessor(framer).
	*
	* The methods are called in a IO thread when the framer received a frame for
	* this stream.
	*
	* They should not block.
	*
	* The frame must be either consumed or popInFrame must be called, after the
	* call is done the frame will be reused.
	*/
	public class SpdyStream implements Runnable {
	private final SpdyConnection spdy;

	public SpdyFrame reqFrame;

	public SpdyFrame resFrame;

	/**
	* For blocking support.
	*/
	private final BlockingQueue<SpdyFrame> inData = new LinkedBlockingQueue<>();

	protected boolean finSent;

	protected boolean finRcvd;

	/**
	* Dummy data frame to insert on reset / go away
	*/
	private static final SpdyFrame END_FRAME;

	static {
	END_FRAME = new SpdyFrame(16);
	END_FRAME.endData = 0;
	END_FRAME.off = 0;
	END_FRAME.c = false;
	END_FRAME.flags =SpdyConnection.FLAG_HALF_CLOSE;
	}

	public SpdyStream(SpdyConnection spdy) {
	this.spdy = spdy;
	}

	public void dump(PrintWriter out) {
	if (reqFrame != null) {
	out.println("Req: " + reqFrame);
	}
	if (resFrame != null) {
	out.println("Res: " + resFrame);
	}
	out.println("In: " + inData.size() + (finRcvd ? " FIN":""));
	}

	/**
	* Non-blocking, called when a data frame is received.
	*/
	public void onDataFrame(SpdyFrame inFrame) {
	synchronized(this) {
	inData.add(inFrame);
	if (inFrame.closed()) {
	finRcvd = true;
	}
	}
	}

	/**
	* Non-blocking - handles a syn stream package. The processor must consume
	* frame.data or set it to null.
	*
	* The base method is for client implementation - servers need to override
	* and process the frame as a request.
	*/
	public void onCtlFrame(SpdyFrame frame) {
	// TODO: handle RST
	if (frame.type == SpdyConnection.TYPE_SYN_STREAM) {
	reqFrame = frame;
	} else if (frame.type == SpdyConnection.TYPE_SYN_REPLY) {
	resFrame = frame;
	}
	synchronized (this) {
	inData.add(frame);
	if (frame.isHalfClose()) {
	finRcvd = true;
	}
	}
	}

	/**
	* Called on GOAWAY or reset.
	*/
	public void onReset() {
	finRcvd = true;
	finSent = true;

	// To unblock
	inData.add(END_FRAME);
	}

	/**
	* True if the channel both received and sent FIN frames.
	*
	* This is tracked by the processor, to avoid extra storage in framer.
	*/
	public boolean isFinished() {
	return finSent && finRcvd;
	}

	/**
	* Waits and return the next data frame, null on timeout.
	*/
	public SpdyFrame getDataFrame(long to) throws IOException {
	while (true) {
	SpdyFrame res = getFrame(to);
	if (res == null \|\| res.isData()) {
	return res;
	}
	if (res.type == SpdyConnection.TYPE_RST_STREAM) {
	throw new IOException("Reset");
	}
	}
	}

	/**
	* Waits and return the next frame.
	*
	* First frame will be the control frame
	*/
	public SpdyFrame getFrame(long to) {
	SpdyFrame in;
	try {
	synchronized (this) {
	if (inData.size() == 0 && finRcvd) {
	return END_FRAME;
	}
	}
	in = inData.poll(to, TimeUnit.MILLISECONDS);
	return in;
	} catch (InterruptedException e) {
	return null;
	}
	}

	public void getHeaders(Map<String, String> resHeaders) {
	SpdyFrame f = resFrame;
	int nvCount = f.nvCount;
	for (int i = 0; i < nvCount; i++) {
	int len = f.read16();
	String n = new String(f.data, f.off, len, StandardCharsets.UTF_8);
	f.advance(len);
	len = f.read16();
	String v = new String(f.data, f.off, len, StandardCharsets.UTF_8);
	f.advance(len);
	resHeaders.put(n, v);
	}
	}

	public SpdyFrame getRequest() {
	if (reqFrame == null) {
	reqFrame = spdy.getFrame(SpdyConnection.TYPE_SYN_STREAM);
	}
	return reqFrame;
	}

	public SpdyFrame getResponse() {
	if (resFrame == null) {
	resFrame = spdy.getFrame(SpdyConnection.TYPE_SYN_REPLY);
	resFrame.streamId = reqFrame.streamId;
	}
	return resFrame;
	}

	public synchronized void sendDataFrame(byte[] data, int start,
	int length, boolean close) {

	SpdyFrame oframe = spdy.getDataFrame();

	// Options:
	// 1. wrap the byte[] data, use a separate header[], wait frame sent
	// -> 2 socket writes
	// 2. copy the data to frame byte[] -> non-blocking queue
	// 3. copy the data, blocking drain -> like 1, trade one copy to
	// avoid
	// 1 tcp packet. That's the current choice, seems closer to rest of
	// tomcat

	if (close)
	oframe.halfClose();

	oframe.append(data, start, length);
	spdy.send(oframe, this);
	}

	public void send() {
	send("http", "GET");
	}

	public void send(String host, String url, String scheme, String method) {
	getRequest().addHeader("host", host);
	getRequest().addHeader("url", url);

	send(scheme, method);
	}

	public void send(String scheme, String method) {
	getRequest();
	if ("GET".equalsIgnoreCase(method)) {
	// TODO: add the others
	reqFrame.halfClose();
	}
	getRequest().addHeader("scheme", scheme);
	getRequest().addHeader("method", method);
	getRequest().addHeader("version", "HTTP/1.1");
	if (reqFrame.isHalfClose()) {
	finSent = true;
	}
	spdy.send(reqFrame, this);
	}

	@Override
	public void run() {
	try {
	spdy.spdyContext.handler.onStream(spdy, this);
	} catch (IOException e) {
	e.printStackTrace();
	// TODO: send rst, error processing the stream.
	}
	}


	public InputStream getInputStream() {
	return new SpdyInputStream();
	}

	private class SpdyInputStream extends InputStream {
	private SpdyFrame current = null;
	private long to = 10000; // TODO

	private void fill() {
	if (current == null \|\| current.off == current.endData) {
	current = getFrame(to);
	}
	}

	@Override
	public int read() throws IOException {
	fill();
	if (current == null) {
	return -1;
	}
	return current.readByte();
	}

	@Override
	public int read(byte b[], int off, int len) throws IOException {
	fill();
	if (current == null) {
	return -1;
	}
	// don't wait for next frame
	int rd = Math.min(len, current.endData - current.off);
	System.arraycopy(current.data, current.off, b, off, rd);
	current.off += rd;
	return rd;
	}

	@Override
	public int available() throws IOException {
	return 0;
	}
	@Override
	public void close() throws IOException {
	// send RST if not closed
	}



	}
	}