Mercurial > hg > Applications > TreeVNC
view src/main/java/jp/ac/u_ryukyu/treevnc/TreeRFBProto.java @ 322:345185ef630e
close viewer as soon as push share screen button.
author | oc |
---|---|
date | Sat, 31 Jan 2015 01:45:39 +0900 |
parents | a237c7e3b7ca |
children | cc9d33208f34 |
line wrap: on
line source
package jp.ac.u_ryukyu.treevnc; import java.io.IOException; import java.io.OutputStream; import java.io.UnsupportedEncodingException; import java.net.*; import java.nio.ByteBuffer; import java.nio.ByteOrder; import java.util.Iterator; import java.util.LinkedList; import java.util.concurrent.atomic.AtomicInteger; import java.util.zip.DataFormatException; import java.util.zip.Deflater; import java.util.zip.Inflater; import com.glavsoft.exceptions.TransportException; import com.glavsoft.rfb.client.ClientToServerMessage; import com.glavsoft.rfb.encoding.EncodingType; import com.glavsoft.rfb.encoding.decoder.FramebufferUpdateRectangle; import com.glavsoft.rfb.protocol.Protocol; import com.glavsoft.rfb.protocol.ProtocolContext; import com.glavsoft.transport.Reader; import com.glavsoft.transport.Writer; import com.glavsoft.viewer.ViewerInterface; import com.glavsoft.viewer.swing.ConnectionParams; public class TreeRFBProto { final static int FramebufferUpdateRequest = 3; final static int CheckDelay = 11; protected final static int FramebufferUpdate = 0; protected ProtocolContext context; private int clients = 0; public MulticastQueue<LinkedList<ByteBuffer>> multicastqueue = new MulticastQueue<LinkedList<ByteBuffer>>(); private RequestScreenThread rThread; public int acceptPort = 0; private String myAddress; private long counter = 0; // packet serial number public ServerSocket servSock; private static final int INFLATE_BUFSIZE = 1024 * 100; private Inflater inflater = new Inflater(); private Deflater deflater = new Deflater(); ViewerInterface viewer; private short id; // my tree node id ( = 0 in root ) private TreeVncCommandChannelListener acceptThread; private TreeRootFinderListener getCast; private CreateConnectionParam cp; private short reconnectingId; // Change Server Request to id's node VNC server private TreeVNCNetwork nets = new TreeVNCNetwork(); private TreeVncRootSelectionPanel rootSelectionPanel; private String vncInterface; public LinkedList<TreeVNCNode> nodeList; protected boolean readyReconnect = false; private boolean cuiVersion; private boolean permitChangeScreen = true; private boolean leader; private boolean firstTime = true; private boolean hasViewer = false; private boolean reconnecting; private boolean normalTermination; private boolean isTreeManager; public boolean showTreeNode = false; public boolean checkDelay = false; public boolean addSerialNum = false; public boolean fixingSize = false; public boolean filterSingleDisplay = false; public int frameSizeWidth; public int frameSizeHeight; public int fixingSizeWidth; public int fixingSizeHeight; private int singleWidth = 0; private int singleHeight = 0; private DatagramSocket socket = null; private byte[] originalInitData = null; public TreeRFBProto(boolean isTreeManager) { rThread = new RequestScreenThread(this); nets.setMyRfb(this); this.isTreeManager = isTreeManager; // this.socket = createSocket(); } public boolean isTreeManager() { return isTreeManager; } public ProtocolContext getContext() { return context; } /** * handle new client accept * it also handle TreeVNC Command * @param newCli * @param os * @param is * @throws IOException * @throws TransportException */ public void newClient(final Socket newCli,final Writer os, final Reader is) { final int myId = clients; final MulticastQueue.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 = 50000 / 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, "timer-discard-multicastqueue").start(); /** * handle command from lower node */ final Runnable reader = new Runnable() { public void run() { for (;;) { try { final byte b[] = new byte[4096]; final int c = is.readByte(b); if (c <= 0) throw new IOException(); // case of root node. if (isTreeManager()) { if (b[0] == ClientToServerMessage.SERVER_CHANGE_REQUEST) { if (permitChangeScreen()) { ByteBuffer buf = ByteBuffer.wrap(b); buf.order(ByteOrder.BIG_ENDIAN); buf.get(); buf.get(); short id = buf.getShort(); int length = buf.getInt(); if (length == 0) continue; byte[] byteAddress = new byte[length]; buf.get(byteAddress); String newHostName = new String(byteAddress, "UTF-8"); int frameSizeWidth = buf.getInt(); int frameSizeHeight = buf.getInt(); System.out.println("Root server change request :" + newHostName); // please remove these numbers. if (viewer != null) { changeVNCServer(viewer, newHostName, frameSizeWidth, frameSizeHeight, id); viewer.setFitScreen(); } } else { continue; } } else if (b[0] == ClientToServerMessage.REQUEST_SINGLE_DISPLAY_SIZE) { ByteBuffer buf = ByteBuffer.wrap(b); buf.order(ByteOrder.BIG_ENDIAN); buf.get(); sendSingleDisplaySize(singleWidth, singleHeight); } else if (b[0] == ClientToServerMessage.READY_SHARE_SOUND) { System.out.println("rootは受信準備を始める。"); int rtpPort = selectPort(ConnectionParams.DEFAULT_RTP_PORT); InetAddress hostname = InetAddress.getLocalHost(); ReceiveSound receiveSound = new ReceiveSound(socket); Thread receiveSoundThread = new Thread(receiveSound, "receive-sound"); receiveSoundThread.start(); } else if (b[0] == ClientToServerMessage.CHECK_DELAY_REPLY) { ByteBuffer buf = ByteBuffer.wrap(b); buf.order(ByteOrder.BIG_ENDIAN); buf.getShort(); Long time = buf.getLong(); int port = buf.getInt(); int addressLength = buf.getInt(); byte[] byteAddress = new byte[addressLength]; buf.get(byteAddress); String address = new String(byteAddress, "UTF-8"); int nodeNum = getNodeNum(port, address); Long delay = System.currentTimeMillis() - time; double halfDelay = (double) delay / 2; System.out.println(System.currentTimeMillis() + " : receive checkDelay : nodeNum" + nodeNum + ", port : " + port + ", address : " + address + ", delay : " + halfDelay); } // case of client node. } else if (b[0] == ClientToServerMessage.SERVER_CHANGE_REQUEST) { ClientToServerMessage serverChangeRequest = new ClientToServerMessage() { @Override public void send(Writer writer) throws TransportException { writer.write(b,0,c); } }; context.sendMessage(serverChangeRequest); } else if (b[0] == ClientToServerMessage.REQUEST_SINGLE_DISPLAY_SIZE) { ClientToServerMessage requestSingeDisplayWidth = new ClientToServerMessage() { @Override public void send(Writer writer) throws TransportException { writer.write(b, 0, c); } }; context.sendMessage(requestSingeDisplayWidth); } else if (b[0] == ClientToServerMessage.READY_SHARE_SOUND) { ClientToServerMessage readyShareSound = new ClientToServerMessage() { @Override public void send(Writer writer) throws TransportException { writer.write(b,0,c); } }; context.sendMessage(readyShareSound); } else if (b[0] == ClientToServerMessage.CHECK_DELAY_REPLY) { ClientToServerMessage checkDelayReply = new ClientToServerMessage() { @Override public void send(Writer writer) throws TransportException { writer.write(b,0,c); } }; context.sendMessage(checkDelayReply); } } catch (Exception e) { try { writerRunning.set(2); os.close(); is.close(); break; } catch (IOException e1) { } catch (TransportException e1) { e1.printStackTrace(); } return; } } } }; /** * send packets to a client (one thread for each client ) */ Runnable sender = new Runnable() { public void run() { writerRunning.set(1); try { requestThreadNotify(); // after this, we discard upward packet. new Thread(reader, "upward-packet-processing").start(); for (;;) { LinkedList<ByteBuffer> bufs = c.poll(); int inputIndex = 0; ByteBuffer header = bufs.get(inputIndex); if (header == null) continue; writeToClient(os, bufs, inputIndex); writerRunning.set(1); // yes my client is awaking. } } catch (Exception e) { try { writerRunning.set(2); os.close(); } catch (IOException e1) { System.out.println("root writer close faild :" + e1); } System.out.println("root writer faild :" + e); /* if socket closed cliList.remove(newCli); */ } } public void writeToClient(final Writer os, LinkedList<ByteBuffer> bufs, int inputIndex) throws TransportException { while (inputIndex < bufs.size()) { ByteBuffer b = bufs.get(inputIndex++); os.write(b.array(), b.position(), b.limit()); } os.flush(); bufs = null; multicastqueue.heapAvailable(); } }; clients++; new Thread(sender, "writer-to-lower-node").start(); } public void sendSingleDisplaySize(int singleDisplayWidth, int singleDisplayHeight) throws UnsupportedEncodingException { LinkedList<ByteBuffer> linkedListSendSingleDisplaySize = new LinkedList<ByteBuffer>(); SendSingleDisplaySize sendSingleDisplaySize = new SendSingleDisplaySize(); linkedListSendSingleDisplaySize.add(sendSingleDisplaySize.singleDisplaySize(0, 0, singleDisplayWidth, singleDisplayHeight)); addSerialNumber(linkedListSendSingleDisplaySize); multicastqueue.put(linkedListSendSingleDisplaySize); System.out.println(singleDisplayWidth + " : send single display width"); } public boolean permitChangeScreen() { return permitChangeScreen; } public void setPermitChangeScreen(boolean v) { permitChangeScreen = v; } public void requestThreadNotify() { rThread.reStart(); } public void setProtocolContext(Protocol workingProtocol) { context = workingProtocol; } public Socket accept() throws IOException { return servSock.accept(); } public int selectPort(int p) { int port = p; while (true) { try { servSock = new ServerSocket(port); acceptPort = port; myAddress = "127.0.0.1"; nets.getNetworkInterfaces(); break; } catch (BindException e) { port++; continue; } catch (IOException e) { } } System.out.println("accept port = " + port); return port; } public void writeFramebufferUpdateRequest(int x, int y, int w, int h, boolean incremental) throws TransportException { byte[] b = new byte[10]; b[0] = (byte) FramebufferUpdateRequest; // 3 is FrameBufferUpdateRequest b[1] = (byte) (incremental ? 1 : 0); b[2] = (byte) ((x >> 8) & 0xff); b[3] = (byte) (x & 0xff); b[4] = (byte) ((y >> 8) & 0xff); b[5] = (byte) (y & 0xff); b[6] = (byte) ((w >> 8) & 0xff); b[7] = (byte) (w & 0xff); b[8] = (byte) ((h >> 8) & 0xff); b[9] = (byte) (h & 0xff); } public void setViewer(ViewerInterface v) { viewer = v; } public ViewerInterface getViewer() { return viewer; } void sendInitData(OutputStream os) throws IOException { os.write(context.getInitData()); } public void setTerminationType(boolean setType) { normalTermination = setType; } public boolean getTerminationType() { return normalTermination; } public void addHostToSelectionPanel(int port, String hostname,String myHostName) { if (rootSelectionPanel != null) { rootSelectionPanel.checkBox(Integer.toString(port) + ":" + hostname + ":" + myHostName); rootSelectionPanel.setButton(); rootSelectionPanel.visible(); } } public void createRootSelectionPanel(CreateConnectionParam cp) { rootSelectionPanel = new TreeVncRootSelectionPanel(); rootSelectionPanel.setCp(cp); } public void close() { // none } public int getAcceptPort() { return acceptPort; } public boolean getReadyReconnect() { return readyReconnect; } public boolean getCuiVersion() { return cuiVersion; } public void setCuiVersion(boolean flag) { cuiVersion = flag; } public void readCheckDelay(Reader reader) throws TransportException { } public synchronized void vncConnected(boolean ready) { if (ready && reconnecting) { Socket vncSocket = viewer.getVNCSocket(); NetworkInterface ni = nets.getInterface(vncSocket); if (ni!=null) { vncInterface = ni.getName(); System.out.println("VNCNetworkInterface :" + vncInterface); } sendDesktopSizeChange(reconnectingId); reconnecting = false; if (reconnectingId == 0) { viewer.setVisible(false); } } if (reconnectingId!=0) { readyReconnect = ready; if (ready) { notifyAll(); } } } public synchronized void waitForVNCConnection() throws InterruptedException { if (reconnectingId!=0) { while (!readyReconnect) { wait(); } } } public void sendDesktopSizeChange(short id) { LinkedList<ByteBuffer> desktopSize = new LinkedList<ByteBuffer>(); int width = context.getFbWidth(); int height = context.getFbHeight(); byte[] initData = null; if (filterSingleDisplay) { this.originalInitData = createOriginalInitData(singleWidth, singleHeight, context.getRemoteDesktopName()); desktopSize.add(new ChengeDesktopSize(singleWidth, singleHeight, EncodingType.INIT_DATA, originalInitData, id).getMessage()); } else { desktopSize.add(new ChengeDesktopSize(width, height, EncodingType.INIT_DATA, context.getInitData(), id).getMessage()); } addSerialNumber(desktopSize); multicastqueue.put(desktopSize); } public byte[] createOriginalInitData(int singleWidth, int singleHeight, String remoteDesktopName) { TreeVncCommandChannelListener treeVncCommandChannelListener = new TreeVncCommandChannelListener(this, acceptPort); byte[] originalInitData = treeVncCommandChannelListener.createOriginalInitData(singleWidth, singleHeight, remoteDesktopName); return originalInitData; } public void addSerialNumber(LinkedList<ByteBuffer> bufs) { ByteBuffer serialNum = multicastqueue.allocate(4+8); // addSerialNum flag + SerialNum if (this.addSerialNum) { serialNum.putInt(1); } else { serialNum.putInt(0); } serialNum.putLong(counter++); serialNum.flip(); bufs.addFirst(serialNum); } public void resetDecoder() { context.resetDecoder(); } public void stopReceiverTask() { if (context!=null) context.cleanUpSession(null); // cleanup zlib decoder for new VNCServer if (isTreeManager()) inflater = new Inflater(); } public String getMyAddress() { return myAddress; } /** * 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 = multicastqueue.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 = multicastqueue.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 bytes * byte data[] * @return number of total bytes * @throws IOException */ public int unzip(Inflater inflater, LinkedList<ByteBuffer> inputs, int inputIndex, byte[] bytes, int bufSize) { int position = 0; int limit = bytes.length; 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; try { len0 = inflater.inflate(bytes, position, limit-position); if (len0 > 0) { position += len0; if (position > limit) throw new DataFormatException(); } } catch (DataFormatException e) { e.printStackTrace(); } } while (!inflater.needsInput()); } return position; } /** * read FrameBuffferUpdate. If it is ZLE, make it ZLEE which is self contained compressed packet. * put the packet to the multicastqueue. Then normal rendering engine read the same stream using is.reset(). * @param dataLen * @param reader * @throws TransportException * @throws UnsupportedEncodingException */ public void readSendData(int dataLen, Reader reader, byte[] bytes, FramebufferUpdateRectangle rect) throws TransportException, UnsupportedEncodingException { LinkedList<ByteBuffer> bufs = new LinkedList<ByteBuffer>(); ByteBuffer header = multicastqueue.allocate(16); ByteBuffer serial = multicastqueue.allocate(4+8); if (!isTreeManager()) { reader.readBytes(serial.array(),0,4+8); serial.limit(4+8); } reader.mark(dataLen); reader.readBytes(header.array(), 0, 16); header.limit(16); if (header.get(0) == FramebufferUpdate) { int encoding = header.getInt(12); if (encoding == EncodingType.ZRLE.getId() || encoding == EncodingType.ZLIB.getId()) { // recompress into ZREE // uncompressed result is remain in bytes ByteBuffer len = multicastqueue.allocate(4); reader.readBytes(len.array(), 0, 4); len.limit(4); ByteBuffer inputData = multicastqueue.allocate(dataLen - 20); reader.readBytes(inputData.array(), 0, inputData.capacity()); inputData.limit(dataLen - 20); LinkedList<ByteBuffer> inputs = new LinkedList<ByteBuffer>(); inputs.add(inputData); header.putInt(12, EncodingType.ZRLEE.getId()); // 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>(); try { unzip(inflater, inputs, 0, bytes, INFLATE_BUFSIZE); // dump32(inputs); out.add(ByteBuffer.wrap(bytes)); int len2 = zip(nDeflater, out, 0, bufs); ByteBuffer blen = multicastqueue.allocate(4); blen.putInt(len2); blen.flip(); bufs.addFirst(blen); bufs.addFirst(header); addSerialNumber(bufs); if (filterSingleDisplay) { if (rect.x < singleWidth) { multicastqueue.put(bufs); } } else { multicastqueue.put(bufs); } } catch (IOException e) { throw new TransportException(e); } return; } // ZRLEE is already compressed bufs.add(header); if (dataLen > 16) { ByteBuffer b = multicastqueue.allocate(dataLen - 16); reader.readBytes(b.array(), 0, dataLen - 16); b.limit(dataLen - 16); bufs.add(b); } this.addSerialNumber(bufs); multicastqueue.put(bufs); return; } // 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. } public void setId(short id) { this.id = id; } public short getId() { return id; } public void setMyAddress(String myHostName) { this.myAddress = myHostName; } public void setLeader(boolean leader) { this.leader = leader; } public boolean isLeader() { return leader; } public void setTreeManager(String intf, TreeManagement clients) { nets.setTreeManager(intf, clients); } public TreeManagement getTreeManager(String intf) { TreeManagement tm = nets.getTreeManager(intf); this.nodeList = tm.getList(); return tm; } /** * change VNCServer is called when host change. * * @param vncProxyService * @param hostName * HostAddress * @param width * FrameWidth * @param height * FrameHeight * @param id * @throws InterruptedException */ public void changeVNCServer(ViewerInterface vncProxyService, String hostName, int width, int height, short id) throws UnknownHostException, IOException, InterruptedException { this.frameSizeWidth = width; this.frameSizeHeight = height; singleWidth = frameSizeWidth; singleHeight = frameSizeHeight; if (filterSingleDisplay) { // send single display width to other node. sendSingleDisplaySize(singleWidth, singleHeight); } reconnectingId = id; vncProxyService.inhelitClients(vncProxyService, hostName); // after connecting VNC server, rfb send SEND_INIT_DATA command and wakes me up if necessary reconnecting = true; // stop reader stop stopReceiverTask(); if (reconnectingId!=0) { waitForVNCConnection(); } } /** * Create send sound thread. */ public void createSendSoundThread(TreeRFBProto rfb) throws SocketException, UnknownHostException { SendSound sendSound; // DatagramSocket socket = new DatagramSocket(selectPort(ConnectionParams.DEFAULT_RTP_PORT), InetAddress.getLocalHost()); if(rfb.isTreeManager()) { sendSound = new SendSound(socket, rfb); } else { String ip = "133.13.57.59"; String port = "60004"; sendSound = new SendSound(socket, rfb, ip, port); } Thread sendSoundThread = new Thread(sendSound, "send-sound"); sendSoundThread.start(); } /** * start accepting children * run rootFinderListener if necessary */ public void createConnectionAndStart(ViewerInterface v) { selectPort(ConnectionParams.DEFAULT_VNC_ROOT); startTreeVncCommandListener(); if(isTreeManager() && firstTime) { getCast = new TreeRootFinderListener(v); Thread treeRootFindThread = new Thread(getCast, "tree-root-find-listener"); treeRootFindThread.start(); firstTime = false; } } public void startTreeVncCommandListener() { acceptThread = new TreeVncCommandChannelListener(this, getAcceptPort()); Thread thread = new Thread(acceptThread, "TreeVNC-accept"); thread.start(); } public TreeVncCommandChannelListener getAcceptThread() { return acceptThread; } public void setConnectionParam(CreateConnectionParam createConnectionParam) { cp = createConnectionParam; } public CreateConnectionParam getConnectionParam() { return cp; } public boolean hasViewer() { return hasViewer; } public void setHasViewer(boolean b) { hasViewer = b; } public void setReconnecting(boolean b) { reconnecting = b; } public int getReconnectingId() { return reconnectingId; } public void setShowTree(boolean showTree) { this.showTreeNode = showTree; } public void setCheckDelay(boolean checkDelay) { this.checkDelay = checkDelay; } public void setAddSerialNum(boolean addSerialNum) { this.addSerialNum = addSerialNum; } public int getNodeNum(int port, String address) { int nodeNum = 0; for (Iterator<TreeVNCNode> i = nodeList.iterator(); i.hasNext();) { TreeVNCNode tvn = (TreeVNCNode)i.next(); if (port == tvn.port && address.equals(tvn.hostname)) { nodeNum = tvn.treeNum; return nodeNum; } } return nodeNum; } public void setFixingSize(boolean fixingSize) { this.fixingSize = fixingSize; } public void setFilterSingleDisplay(boolean filterSingleDisplay) { this.filterSingleDisplay = filterSingleDisplay; } private DatagramSocket createSocket() { try { if (socket == null) { this.socket = new DatagramSocket(ConnectionParams.DEFAULT_RTP_PORT, InetAddress.getLocalHost()); } } catch (SocketException e) { e.printStackTrace(); } catch (UnknownHostException e) { e.printStackTrace(); } return socket; } public void setSingleDisplaySize(int singleWidth, int singleHeight) { this.singleWidth = singleWidth; this.singleHeight = singleHeight; } public int getSingleWidth() { return singleWidth; } public int getSingleHeight() { return singleHeight; } }