Mercurial > hg > Members > nobuyasu > tightVNCProxy
view src/myVncProxy/MyRfbProto.java @ 138:001f0b770f96
modify MyRfbProto
author | e085711 |
---|---|
date | Tue, 30 Aug 2011 19:59:02 +0900 |
parents | e97c0f82eaf0 |
children | fc77596d3064 0d9a824dd12f |
line wrap: on
line source
package myVncProxy; import static org.junit.Assert.*; import java.awt.Graphics; import java.awt.Image; import java.awt.image.BufferedImage; import java.io.BufferedOutputStream; import java.io.BufferedReader; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.InputStream; import java.io.InputStreamReader; import java.net.BindException; import java.net.ServerSocket; import java.net.Socket; import java.nio.ByteBuffer; import java.util.Iterator; import java.util.LinkedList; import javax.imageio.ImageIO; import org.junit.Test; import myVncProxy.MulticastQueue.Client; import java.util.concurrent.ExecutorService; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.zip.DataFormatException; import java.util.zip.Deflater; import java.util.zip.Inflater; import java.io.OutputStream; public class MyRfbProto extends RfbProto { final static String versionMsg_3_998 = "RFB 003.998\n"; /** * CheckMillis is one of new msgType for RFB 3.998. */ final static byte SpeedCheckMillis = 4; private static final int INFLATE_BUFSIZE = 1024*100; boolean printStatusFlag = false; long startCheckTime; private int messageType; private int rectangles; private int rectX; private int rectY; private int rectW; private int rectH; private int encoding; private int zLen; private boolean clicomp = false; private ServerSocket servSock; private int acceptPort; private byte initData[]; private LinkedList<Socket> cliListTmp; private LinkedList<Socket> cliList; boolean createBimgFlag; ExecutorService executor; byte[] pngBytes; // private MulticastQueue<LinkedList<ByteBuffer>> multicastqueue = new MostRecentMultiCast<LinkedList<ByteBuffer>>(10); private MulticastQueue<LinkedList<ByteBuffer>> multicastqueue = new MulticastQueue<LinkedList<ByteBuffer>>(); private int clients = 0; private Inflater inflater = new Inflater(); private Deflater deflater = new Deflater(); private CreateThread geth; public MyRfbProto() throws IOException { } MyRfbProto(String h, int p, VncViewer v) throws IOException { super(h, p, v); } MyRfbProto(String h, int p, CreateThread geth) throws IOException { super(h, p); this.geth = geth; } MyRfbProto(String h, int p) throws IOException { super(h, p); } // over write void writeVersionMsg() throws IOException { clientMajor = 3; if (serverMinor >= 9) { clientMinor = 9; os.write(versionMsg_3_998.getBytes()); } else if (serverMajor > 3 || serverMinor >= 8) { clientMinor = 8; os.write(versionMsg_3_8.getBytes()); } else if (serverMinor >= 9) { clientMinor = 9; os.write(versionMsg_3_998.getBytes()); } else if (serverMinor >= 7) { clientMinor = 7; os.write(versionMsg_3_7.getBytes()); } else { clientMinor = 3; os.write(versionMsg_3_3.getBytes()); } protocolTightVNC = false; initCapabilities(); } void initServSock(int port) throws IOException { servSock = new ServerSocket(port); acceptPort = port; } // 5999を開けるが、開いてないなら+1のポートを開ける。 void selectPort(int p) { int port = p; while (true) { try { initServSock(port); break; } catch (BindException e) { port++; continue; } catch (IOException e) { } } System.out.println("accept port = " + port); } int getAcceptPort() { return acceptPort; } void setSoTimeout(int num) throws IOException { servSock.setSoTimeout(num); } Socket accept() throws IOException { return servSock.accept(); } void addSock(Socket sock) { cliList.add(sock); } void addSockTmp(Socket sock) { System.out.println("connected " + sock.getInetAddress()); cliListTmp.add(sock); } boolean markSupported() { return is.markSupported(); } void readServerInit() throws IOException { is.mark(255); skipBytes(20); int nlen = readU32(); int blen = 20 + 4 + nlen; initData = new byte[blen]; is.reset(); is.mark(blen); readFully(initData); is.reset(); framebufferWidth = readU16(); framebufferHeight = readU16(); bitsPerPixel = readU8(); depth = readU8(); bigEndian = (readU8() != 0); trueColour = (readU8() != 0); redMax = readU16(); greenMax = readU16(); blueMax = readU16(); redShift = readU8(); greenShift = readU8(); blueShift = readU8(); byte[] pad = new byte[3]; readFully(pad); int nameLength = readU32(); byte[] name = new byte[nameLength]; readFully(name); desktopName = new String(name); // Read interaction capabilities (TightVNC protocol extensions) if (protocolTightVNC) { int nServerMessageTypes = readU16(); int nClientMessageTypes = readU16(); int nEncodingTypes = readU16(); readU16(); readCapabilityList(serverMsgCaps, nServerMessageTypes); readCapabilityList(clientMsgCaps, nClientMessageTypes); readCapabilityList(encodingCaps, nEncodingTypes); } inNormalProtocol = true; } void sendRfbVersion(OutputStream os) throws IOException { os.write(versionMsg_3_998.getBytes()); // os.write(versionMsg_3_8.getBytes()); } void readVersionMsg(InputStream is, OutputStream os) throws IOException { byte[] b = new byte[12]; is.read(b); if ((b[0] != 'R') || (b[1] != 'F') || (b[2] != 'B') || (b[3] != ' ') || (b[4] < '0') || (b[4] > '9') || (b[5] < '0') || (b[5] > '9') || (b[6] < '0') || (b[6] > '9') || (b[7] != '.') || (b[8] < '0') || (b[8] > '9') || (b[9] < '0') || (b[9] > '9') || (b[10] < '0') || (b[10] > '9') || (b[11] != '\n')) { throw new IOException("Host " + host + " port " + port + " is not an RFB server"); } serverMajor = (b[4] - '0') * 100 + (b[5] - '0') * 10 + (b[6] - '0'); serverMinor = (b[8] - '0') * 100 + (b[9] - '0') * 10 + (b[10] - '0'); if (serverMajor < 3) { throw new IOException( "RFB server does not support protocol version 3"); } if (serverMinor == 998) { sendPortNumber(os); } } void sendPortNumber(OutputStream os) throws IOException { byte[] b = new byte[4]; b = castIntByte(geth.port); os.write(b); } void sendSecurityType(OutputStream os) throws IOException { // number-of-security-types os.write(1); // security-types // 1:None os.write(1); } void readSecType(InputStream is) throws IOException { byte[] b = new byte[1]; is.read(b); } void sendSecResult(OutputStream os) throws IOException { byte[] b = castIntByte(0); os.write(b); } void readClientInit(InputStream in) throws IOException { byte[] b = new byte[0]; in.read(b); } void sendInitData(OutputStream os) throws IOException { os.write(initData); } void sendPngImage() { try { for (Socket cli : cliListTmp) { try { sendPngData(cli); addSock(cli); } catch (IOException e) { // if socket closed cliListTmp.remove(cli); } } // System.out.println("cliSize="+cliSize()); } catch (Exception e) { } cliListTmp.clear(); } boolean ready() throws IOException { BufferedReader br = new BufferedReader(new InputStreamReader(is)); return br.ready(); } int cliSize() { return cliList.size(); } void printNumBytesRead() { System.out.println("numBytesRead=" + numBytesRead); } void regiFramebufferUpdate() throws IOException { is.mark(20); messageType = readU8(); // 0 skipBytes(1); // 1 rectangles = readU16(); // 2 rectX = readU16(); // 4 rectY = readU16(); // 6 rectW = readU16(); // 8 rectH = readU16(); // 10 encoding = readU32(); // 12 // System.out.println("encoding = "+encoding); if (encoding == EncodingZRLE|| encoding==EncodingZRLEE||encoding==EncodingZlib) zLen = readU32(); else zLen = 0; is.reset(); } int checkAndMark() throws IOException { int dataLen; switch (encoding) { case RfbProto.EncodingRaw: dataLen = rectW * rectH * 4 + 16; // is.mark(dataLen); break; case RfbProto.EncodingCopyRect: dataLen = 16 + 4; // is.mark(dataLen); break; case RfbProto.EncodingRRE: case RfbProto.EncodingCoRRE: case RfbProto.EncodingHextile: case RfbProto.EncodingTight: dataLen = zLen + 20; // is.mark(dataLen); break; case RfbProto.EncodingZlib: case RfbProto.EncodingZRLE: case RfbProto.EncodingZRLEE: dataLen = zLen + 20; // is.mark(dataLen); break; case RfbProto.EncodingXCursor: case RfbProto.EncodingRichCursor: int pixArray = rectW * rectH * 4; int u8Array = (int)Math.floor((rectW + 7)/8) * rectH; dataLen = pixArray + u8Array; printFramebufferUpdate(); // is.mark(dataLen); break; default: dataLen = 1000000; // is.mark(dataLen); } return dataLen; } void sendDataToClient() throws Exception { regiFramebufferUpdate(); // printFramebufferUpdate(); int dataLen = checkAndMark(); readSendData(dataLen); } BufferedImage createBufferedImage(Image img) { BufferedImage bimg = new BufferedImage(img.getWidth(null), img.getHeight(null), BufferedImage.TYPE_INT_RGB); Graphics g = bimg.getGraphics(); g.drawImage(img, 0, 0, null); g.dispose(); return bimg; } void createPngBytes(BufferedImage bimg) throws IOException { pngBytes = getImageBytes(bimg, "png"); } byte[] getBytes(BufferedImage img) throws IOException { byte[] b = getImageBytes(img, "png"); return b; } byte[] getImageBytes(BufferedImage image, String imageFormat) throws IOException { ByteArrayOutputStream bos = new ByteArrayOutputStream(); BufferedOutputStream os = new BufferedOutputStream(bos); image.flush(); ImageIO.write(image, imageFormat, os); os.flush(); os.close(); return bos.toByteArray(); } void sendPngData(Socket sock) throws IOException { byte[] dataLength = castIntByte(pngBytes.length); sock.getOutputStream().write(dataLength); sock.getOutputStream().write(pngBytes); } byte[] castIntByte(int len) { byte[] b = new byte[4]; b[0] = (byte) ((len >>> 24) & 0xFF); b[1] = (byte) ((len >>> 16) & 0xFF); b[2] = (byte) ((len >>> 8) & 0xFF); b[3] = (byte) ((len >>> 0) & 0xFF); return b; } BufferedImage createBimg() throws IOException { BufferedImage bimg = ImageIO.read(new ByteArrayInputStream(pngBytes)); return bimg; } void printFramebufferUpdate() { System.out.println("messageType=" + messageType); System.out.println("rectangles=" + rectangles); System.out.println("encoding=" + encoding); System.out.println("rectX = "+rectX+": rectY = "+rectY); System.out.println("rectW = "+rectW+": rectH = "+rectH); switch (encoding) { case RfbProto.EncodingRaw: System.out.println("rectW * rectH * 4 + 16 =" + rectW * rectH * 4 + 16); break; default: } } int returnMsgtype() { return messageType; } void readSpeedCheck() throws IOException { byte[] b = new byte[1]; readFully(b); } void startSpeedCheck() { ByteBuffer b = ByteBuffer.allocate(10); b.put((byte)SpeedCheckMillis); b.flip(); startCheckTime = System.currentTimeMillis(); System.out.println("startChckTime = "+ startCheckTime); LinkedList<ByteBuffer>bufs = new LinkedList<ByteBuffer>(); bufs.add(b); multicastqueue.put(bufs); } void endSpeedCheck() { long accTime = System.currentTimeMillis(); long time = accTime - startCheckTime; System.out.println("checkMillis: " + time); } synchronized void changeStatusFlag() { printStatusFlag = true; } void printMills() { if(printStatusFlag) { changeStatusFlag(); } else { changeStatusFlag(); } } void speedCheckMillis() { Runnable stdin = new Runnable() { public void run() { int c; try { while( (c = System.in.read()) != -1 ) { switch(c) { case 's': break; default: startSpeedCheck(); break; } } }catch(IOException e){ System.out.println(e); } } }; new Thread(stdin).start(); } /** * gzip byte arrays * @param deflater * @param inputs * byte data[] * @param inputIndex * @param outputs * byte data[] * @return byte length in last byte array * @throws IOException */ public int zip(Deflater deflater,LinkedList<ByteBuffer> inputs, int inputIndex, LinkedList<ByteBuffer> outputs) throws IOException { int len = 0; ByteBuffer c1= ByteBuffer.allocate(INFLATE_BUFSIZE); while(inputIndex < inputs.size() ) { ByteBuffer b1 = inputs.get(inputIndex++); deflater.setInput(b1.array(),b1.position(),b1.remaining()); /** * If we finish() stream and reset() it, Deflater start new gzip stream, this makes continuous zlib reader unhappy. * if we remove finish(), Deflater.deflate() never flushes its output. The original zlib deflate has flush flag. I'm pretty * sure this a kind of bug of Java library. */ if (inputIndex==inputs.size()) deflater.finish(); int len1 = 0; do { len1 = deflater.deflate(c1.array(),c1.position(),c1.remaining()); if (len1>0) { len += len1; c1.position(c1.position()+len1); if (c1.remaining()==0) { c1.flip(); outputs.addLast(c1); c1 = ByteBuffer.allocate(INFLATE_BUFSIZE); } } } while (len1 >0 || !deflater.needsInput()); // &&!deflater.finished()); } if (c1.position()!=0) { c1.flip(); outputs.addLast(c1); } deflater.reset(); return len; } /** * gunzip byte arrays * @param inflater * @param inputs * byte data[] * @param outputs * byte data[] *@return number of total bytes * @throws IOException */ public int unzip(Inflater inflater, LinkedList<ByteBuffer> inputs, int inputIndex, LinkedList<ByteBuffer> outputs,int bufSize) throws DataFormatException { int len=0; ByteBuffer buf = ByteBuffer.allocate(bufSize); while (inputIndex < inputs.size()) { ByteBuffer input = inputs.get(inputIndex++); inflater.setInput(input.array(),input.position(),input.limit()); // if (inputIndex==inputs.size()) if inflater/deflater has symmetry, we need this // inflater.end(); but this won't work do { int len0 = inflater.inflate(buf.array(),buf.position(),buf.remaining()); if (len0>0) { buf.position(buf.position()+len0); len += len0; if (buf.remaining()==0) { buf.flip(); outputs.addLast(buf); buf = ByteBuffer.allocate(bufSize); } } } while (!inflater.needsInput()); } if (buf.position()!=0) { buf.flip(); outputs.addLast(buf); } return len; } /** * send data to clients * @param dataLen * @throws IOException * @throws DataFormatException * * Zlibed packet is compressed in context dependent way, that is, it have to send from the beginning. But this is * impossible. So we have to compress it again for each clients. Separate deflater for each clients is necessary. * * Java's deflater does not support flush. This means to get the result, we have to finish the compression. Reseting * start new compression, but it is not accepted well in zlib continuous reading. So we need new Encoding ZRLEE * which reset decoder for each packet. ZRLEE can be invisible from user, but it have to be implemented in the clients. * ZRLEE compression is not context dependent, so no recompression is necessary. */ void readSendData(int dataLen) throws IOException, DataFormatException { LinkedList<ByteBuffer>bufs = new LinkedList<ByteBuffer>(); ByteBuffer header = ByteBuffer.allocate(16); readFully(header.array(),0,16); header.limit(16); if (header.get(0)==RfbProto.FramebufferUpdate) { int encoding = header.getInt(12); if (encoding==RfbProto.EncodingZRLE||encoding==RfbProto.EncodingZlib) { // ZRLEE is already recompressed ByteBuffer len = ByteBuffer.allocate(4); readFully(len.array(),0,4); len.limit(4); ByteBuffer inputData = ByteBuffer.allocate(dataLen-20); readFully(inputData.array(),0,inputData.capacity()); inputData.limit(dataLen-20); LinkedList<ByteBuffer>inputs = new LinkedList<ByteBuffer>(); inputs.add(inputData); header.putInt(12, RfbProto.EncodingZRLEE); // means recompress every time // using new Deflecter every time is incompatible with the protocol, clients have to be modified. Deflater nDeflater = deflater; // new Deflater(); LinkedList<ByteBuffer> out = new LinkedList<ByteBuffer>(); unzip(inflater, inputs, 0 , out, INFLATE_BUFSIZE); // dump32(inputs); int len2 = zip(nDeflater, out, 0, bufs); ByteBuffer blen = ByteBuffer.allocate(4); blen.putInt(len2); blen.flip(); bufs.addFirst(blen); bufs.addFirst(header); multicastqueue.put(bufs); // is.reset(); return ; } bufs.add(header); if (dataLen>16) { ByteBuffer b = ByteBuffer.allocate(dataLen-16); readFully(b.array(),0,dataLen-16); b.limit(dataLen-16); bufs.add(b); } multicastqueue.put(bufs); // is.reset(); return ; } is.reset(); // It may be compressed. We can inflate here to avoid repeating clients decompressing here, // but it may generate too many large data. It is better to do it in each client. // But we have do inflation for all input data, so we have to do it here. } void newClient(AcceptThread acceptThread, final Socket newCli, final OutputStream os, final InputStream is) throws IOException { // createBimgFlag = true; // rfb.addSockTmp(newCli); // addSock(newCli); final int myId = clients; final Client <LinkedList<ByteBuffer>> c = multicastqueue.newClient(); final AtomicInteger writerRunning = new AtomicInteger(); writerRunning.set(1); /** * Timeout thread. If a client is suspended, it has top of queue indefinitely, which caused memory * overflow. After the timeout, we poll the queue and discard it. Start long wait if writer is running. */ final Runnable timer = new Runnable() { public void run() { int count = 0; for(;;) { long timeout = 30000/8; try { synchronized(this) { int state,flag; writerRunning.set(0); wait(timeout); flag = 0; while((state=writerRunning.get())==0) { c.poll(); // discard, should be timeout count++; if (flag==0) { System.out.println("Discarding "+myId + " count="+ count); flag = 1; } wait(10); // if this is too short, writer cannot take the poll, if this is too long, memory will overflow... } if (flag==1) System.out.println("Resuming "+myId + " count="+count); if (state!=1) { System.out.println("Client died "+myId); break; } } } catch (InterruptedException e) { } } } }; new Thread(timer).start(); /** * discard all incoming from clients */ final Runnable reader = new Runnable() { public void run() { byte b[] = new byte[4096]; for(;;) { try { int c = is.read(b); if (c<=0) throw new IOException(); // System.out.println("client read "+c); } catch (IOException e) { try { writerRunning.set(2); os.close(); is.close(); } catch (IOException e1) { } return; } } } }; /** * send packets to a client */ Runnable sender = new Runnable() { public void run() { writerRunning.set(1); try { /** * initial connection of RFB protocol */ sendRfbVersion(os); // readVersionMsg(is); readVersionMsg(is,os); sendSecurityType(os); readSecType(is); sendSecResult(os); readClientInit(is); sendInitData(os); new Thread(reader).start(); // discard incoming packet here after. for (;;) { LinkedList<ByteBuffer> bufs = c.poll(); int inputIndex = 0; ByteBuffer header = bufs.get(inputIndex); if (header==null) continue; if (header.get(0)==RfbProto.FramebufferUpdate) { // System.out.println("client "+ myId); } writeToClient(os, bufs, inputIndex); writerRunning.set(1); // yes my client is awaking. } } catch (IOException e) { try { writerRunning.set(2); os.close(); } catch (IOException e1) { } /* if socket closed cliList.remove(newCli); */ } } public void writeToClient(final OutputStream os, LinkedList<ByteBuffer> bufs, int inputIndex) throws IOException { while(inputIndex < bufs.size()) { ByteBuffer b = bufs.get(inputIndex++); os.write(b.array(), b.position(), b.limit()); } os.flush(); } }; clients++; new Thread(sender).start(); } public void dump32(LinkedList<ByteBuffer>bufs) { int len =0; for(ByteBuffer b: bufs) len += b.remaining(); ByteBuffer top = bufs.getFirst(); ByteBuffer end = bufs.getLast(); System.err.println("length: "+len); System.err.print("head 0: "); for(int i = 0; i<16 && i < top.remaining(); i++) { System.err.print(" "+ top.get(i)); } System.err.print("tail 0: "); for(int i = 0; i<16 && i < end.remaining(); i++) { System.err.print(" "+end.get(i)); } System.err.println(); } @Test public void test1() { try { LinkedList<ByteBuffer> in = new LinkedList<ByteBuffer>(); LinkedList<ByteBuffer> out = new LinkedList<ByteBuffer>(); LinkedList<ByteBuffer> out2 = new LinkedList<ByteBuffer>(); // if (false) { // for(int i=0;i<10;i++) { // in.add(ByteBuffer.wrap("test1".getBytes())); // in.add(ByteBuffer.wrap("test2".getBytes())); // in.add(ByteBuffer.wrap("test3".getBytes())); // in.add(ByteBuffer.wrap("test44".getBytes())); // } // } else { String t = ""; for(int i=0;i<10;i++) { t += "test1"; t += "test2"; t += "test3"; t += "test44"; } in.add(ByteBuffer.wrap(t.getBytes())); } LinkedList<ByteBuffer> in1 = clone(in); Deflater deflater = new Deflater(); zip(deflater,in,0,out); // LinkedList<ByteBuffer> out3 = clone(out); zipped result is depend on deflator's state unzip(inflater, out, 0,out2, INFLATE_BUFSIZE); // inflater.reset(); equalByteBuffers(in1, out2); LinkedList<ByteBuffer> out4 = new LinkedList<ByteBuffer>(); deflater = new Deflater(); zip(deflater,out2,0,out4); LinkedList<ByteBuffer> out5 = new LinkedList<ByteBuffer>(); unzip(inflater,out4,0, out5, INFLATE_BUFSIZE); int len = equalByteBuffers(in1,out5); System.out.println("Test Ok. "+len); } catch (Exception e) { assertEquals(0,1); } } private LinkedList<ByteBuffer> clone(LinkedList<ByteBuffer> in) { LinkedList<ByteBuffer> copy = new LinkedList<ByteBuffer>(); for(ByteBuffer b: in) { ByteBuffer c = b.duplicate(); copy.add(c); } return copy; } public int equalByteBuffers(LinkedList<ByteBuffer> in, LinkedList<ByteBuffer> out2) { int len = 0; Iterable<Byte> i = byteBufferIterator(in); Iterator<Byte> o = byteBufferIterator(out2).iterator(); for(int b: i) { len ++; if (o.hasNext()) { int c = o.next(); assertEquals(b,c); } else assertEquals(0,1); } if (o.hasNext()) assertEquals(0,1); // System.out.println(); return len; } private Iterable<Byte> byteBufferIterator(final LinkedList<ByteBuffer> in) { return new Iterable<Byte>() { public Iterator<Byte> iterator() { return new Iterator<Byte>() { int bytes = 0; int buffers = 0; public boolean hasNext() { for(;;) { if (buffers>=in.size()) return false; ByteBuffer b = in.get(buffers); if (! (bytes<b.remaining())) { buffers ++; bytes=0; } else return true; } } public Byte next() { ByteBuffer bf =in.get(buffers); byte b = bf.get(bytes++); if (bf.remaining()<=bytes) { buffers++; bytes = 0; } // System.out.print(b); return b; } public void remove() { } }; } }; } }