experserver.java

MANTIS是由科罗拉多大学开发的传感器网络嵌入式操作系统。 这是mantis的0.9.5版本的源码。

            			case AQSHELL_REQUEST:
            			case AQSHELL_DATA:
            			case AQSHELL_NOOP:
                		case AQSHELL_GETID_REPLY:
            			case AQSHELL_SUMMARY_SEND:
            			case AQSHELL_PROFILE_SEND:
            			case AQSHELL_REQUEST_SEND: 
            			case AQSHELL_DATA_SEND:
            			case AQSHELL_SETCACHESIZE:
                            // There is no special handling for these types
            				break;
            			case AQSHELL_COMPLETEPAGE:
                            // Display the new page number
                            if (ui != null)
                                ui.showVersionPage(rpkt.data[0], rpkt.data[1]+1);
                            System.out.print((rpkt.data[1]+1)+" ");
            			case AQSHELL_ALLQUIET:
                            // This node has not transmitted any data or requests for 30 seconds
            				synchronized (this) {
            					if (!allquiet) {
            						allquiet = true;
                                    // Wake up any experiments that may be waiting
            						notify();
            					}
            				}
            				break;
            			case AQSHELL_NOTQUIET:
                            // This node has broken its silence by transmitting a request or data packet
            				synchronized (this) {
            					if (allquiet) {
            						allquiet = false;
                                    // Wake up any experiments that may be waiting
            						notify();
            					}
            				}
            				break;
            			case AQSHELL_NEWVERSION:
                            // This node may be beginning a new update
            				finishTime = 0;
                            if (ui != null) {
                				ui.setFinished(false);
                                ui.showVersionPage(rpkt.data[0], 0);
                            }
            				break;
            			case AQSHELL_COMPLETEPROG:
                            // This node has finished downloading the update
                            if (ui != null)
                                ui.setFinished(true);
                            if (isTarget) {
                				log("Finished %d\n", rpkt.id);
                				log("Time: %10d.%06d\n", rpkt.ts_sec, rpkt.ts_usec);
                            }
                            synchronized (this) {
                                finishTime = 1000L * rpkt.ts_sec + rpkt.ts_usec / 1000;
                                // Notify any experiment that might be listening for this node to finish
                                notify();
                            }
            				break;
            			case AQSHELL_GETSTATS:
            			case AQSHELL_GETSTATS_REPLY: {
            				if ((rpkt.flags & AQSHELL_F_ACK) != 0) {
            					// Final ACK  packet only, no data
								logStats(this);
            					break;
            				}
                            // Copy data into the stats array
            				DataInput di = rpkt.getDataInput();
            	            int row = di.readUnsignedShort();
            	            //System.out.println("Reading rows "+row+" to "+(row+4));
            	            for (int r=row; r<row+4 && r<stats.length; r++)
            	            {
            	                for (int i=0; i<4; i++)
            	                {
            	                    stats[r][i] = di.readUnsignedShort();
            	                }
            	            }
            				break;
            			}
                        case AQSHELL_STOPUPDATE:
                            version = rpkt.data[0];
                            size = rpkt.data[1];
                            if (ui != null)
                                ui.showVersionPage(rpkt.data[0], rpkt.data[1]);
                            break;
            			case AQSHELL_GETVERSION:
                            // Display the current version info for this node
                            version = rpkt.data[0];
                            size = rpkt.data[1];
                            if (ui != null)
                                ui.showVersionPage(rpkt.data[0], rpkt.data[1]);
            				break;
                        case AQSHELL_SETIMAGESIZE:
                            // This node is acknowledging that its image size was updated by aqshell
                            versionTime = 1000L * rpkt.ts_sec + rpkt.ts_usec / 1000;
                            version = rpkt.data[1];
                            size = rpkt.data[0];
                            if (ui != null) {
                                ui.setMessage("Size Ack");
                                ui.showVersionPage(rpkt.data[1], rpkt.data[0]);
                            }
                            log("Version acknowledge %d\n", rpkt.id);
                            log("Time: %10d.%06d\n", rpkt.ts_sec, rpkt.ts_usec);
                            break;
            			case AQSHELL_SETVERSION:
            			case AQSHELL_SETVERSION_REPLY:
                            // This node is acknowledging that its version number was updated by aqshell
            				versionTime = 1000L * rpkt.ts_sec + rpkt.ts_usec / 1000;
                            version = rpkt.data[0];
                            size = rpkt.data[1];
                            if (ui != null) {
                				ui.setMessage("Version Ack");
                                ui.showVersionPage(rpkt.data[0], rpkt.data[1]);
                            }
            				log("Version acknowledge %d\n", rpkt.id);
            				log("Time: %10d.%06d\n", rpkt.ts_sec, rpkt.ts_usec);
            				break;
            			case AQSHELL_CLOSE:
                            // The aqshell process was interrupted, so it is telling us to close the connection
                            // It is cleaner if the client closes a TCP connection first
            				System.out.println("Received AQSHELL_CLOSE");
            				sock.close();
            				break;
                        default:
                            System.err.println("Unrecognized type: "+Integer.toHexString(rpkt.command));
                    }

                	if ((rpkt.flags & AQSHELL_F_ACK) != 0) {
                    	synchronized (this) {
                            // Save the command if this was an ACK
	                		lastAck = rpkt.command;
                            // Notify any experiment that might be listening
	                		notify();
	                	}
                	}
                }
            }
            catch (EOFException e)
            {
                System.err.println(e);
            	//e.printStackTrace();
            }
            catch (IOException e)
            {
                e.printStackTrace();
            }
        	try
			{
                // Close the socket so aqshell knows we disconnected
                // TODO if we are the server, and aqshell is still connected, aqshell should close the socket first
				sock.close();
			}
			catch (IOException e1)
			{
				e1.printStackTrace();
			}
            // Detach from GUI
            if (ui != null)
            	ui.setConnection(null);
            ui = null;
            // Remove ourselves from the list of connections
            connections.remove(this);
            // And ports
            ports.remove(key);
        }

        /* Send a command to the node to change its version number. */
		public void setVersion(int i)
		{
            try
			{
            	out.write(syncbytes);
            	AqShellPacket spkt = new AqShellPacket(AQSHELL_SETVERSION, 
													   AQSHELL_F_PLEASEACK|AQSHELL_F_NODECTL|AQSHELL_F_FORWARDREPLY,
													   -1, 0, 1);
            	spkt.data[0] = (byte)i;
            	spkt.send(out);
			}
			catch (IOException e)
			{
				e.printStackTrace();
			}
		}

        /* Send a command to the node to change its size and version.
         * We have to change the version number at the same time so updates don't occur from two different images. */
		public void setImageSize(int size, int version)
		{
            try
			{
            	out.write(syncbytes);
            	AqShellPacket spkt = new AqShellPacket(AQSHELL_SETIMAGESIZE, 
													   AQSHELL_F_PLEASEACK|AQSHELL_F_NODECTL|AQSHELL_F_FORWARDREPLY,
													   -1, 0, 2);
            	spkt.data[0] = (byte)size;
            	spkt.data[1] = (byte)version;
            	spkt.send(out);
			}
			catch (IOException e)
			{
				e.printStackTrace();
			}
		}

        /* Send a command without any extra data to the node. */
        public void sendNodeCommand(byte command)
		{
            try
			{
            	out.write(syncbytes);
            	AqShellPacket spkt = new AqShellPacket(command, 
													   AQSHELL_F_PLEASEACK|AQSHELL_F_NODECTL|AQSHELL_F_FORWARDREPLY,
													   -1, 0, 0);
            	spkt.send(out);
			}
			catch (IOException e)
			{
				e.printStackTrace();
			}
		}

        /* Send a command without any extra data to aqshell. */
		public void sendShellCommand(byte command)
		{
            try
			{
            	out.write(syncbytes);
            	AqShellPacket spkt = new AqShellPacket(command, 
													   AQSHELL_F_PLEASEACK|AQSHELL_F_SHELLCTL,
													   -1, 0, 0);
            	spkt.send(out);
			}
			catch (IOException e)
			{
				e.printStackTrace();
			}
		}

        /* Set the ACK command.  Used to clear the value by setting it to -1. */
		public synchronized void setAck(int i)
		{
			lastAck = i;
		}

		/* Returns the last command that was acknowledged. */
		public synchronized int getAck()
		{
			return lastAck;
		}

        /* Query a node for its version number and keep resending the query every five seconds. */
		public int getVersion() throws InterruptedException
		{
			synchronized (this) {
				lastAck = -1;
				while (lastAck != AQSHELL_GETVERSION) {
					sendNodeCommand(AQSHELL_GETVERSION);
					wait(5000);
				}
			}
			return version;
		}

        /* Send a command to the node to change its forwarding cache size. 
         * app is the index in src/mos/deluge_proto.c's list of applications. */
		public void setCacheSize(int size, int app)
		{
            try
			{
            	out.write(syncbytes);
            	AqShellPacket spkt = new AqShellPacket(AQSHELL_SETCACHESIZE, 
													   AQSHELL_F_PLEASEACK|AQSHELL_F_NODECTL|AQSHELL_F_FORWARDREPLY,
													   -1, 0, 2);
            	spkt.data[0] = (byte)size;
            	spkt.data[1] = (byte)app;
            	spkt.send(out);
			}
			catch (IOException e)
			{
				e.printStackTrace();
			}
		}

        public void stopUpdating(int app) throws InterruptedException
        {
            try
            {
                synchronized (this) {
                    lastAck = -1;
                    while (lastAck != AQSHELL_STOPUPDATE) {
                        out.write(syncbytes);
                        AqShellPacket spkt = new AqShellPacket(AQSHELL_STOPUPDATE, 
                                                               AQSHELL_F_PLEASEACK|AQSHELL_F_NODECTL|AQSHELL_F_FORWARDREPLY,
                                                               -1, 0, 1);
                        spkt.data[0] = (byte)app;
                        spkt.send(out);
                        wait(5000);
                    }
                }
            }
            catch (IOException e)
            {
                e.printStackTrace();
            }
        }
    }
    
    /* This is a GUI for interacting with the testbed. */
    class ControlPanel extends JPanel implements ActionListener
    {
		private static final long serialVersionUID = 3616444613897107767L;    // Eclipse told me to put this here
        JButton bExperAll;
        JTextField tStartSize;
        JTextField tStartIter;
        
        JButton bExperV;
        JButton bExperS;
        JButton bStop;
        JButton bScanUsb;
        JButton bPoll;
        JButton bPollAll;
        JToggleButton bLogFile;
        JTextField tLogFile;
        ButtonGroup seedSelect;
        
        Thread experThread;
        
        ControlPanel()
        {
            setLayout(new BorderLayout());
            
            JPanel p = new JPanel();
            bExperAll = new JButton("Run All");
            bExperAll.addActionListener(this);
            p.add(bExperAll);
            p.add(new JLabel("Start size:"));
            tStartSize = new JTextField(5);
            tStartSize.setText("2");
            p.add(tStartSize);
            p.add(new JLabel("Start iteration:"));
            tStartIter = new JTextField(5);
            tStartIter.setText("0");
            p.add(tStartIter);

            bExperV = new JButton("Experiment (Bump Version)");
            bExperV.addActionListener(this);
            p.add(bExperV);
            bExperS = new JButton("Experiment (Bump Version and Size)");
            bExperS.addActionListener(this);
            p.add(bExperS);
            bStop = new JButton("Stop");
            bStop.addActionListener(this);
            p.add(bStop);
            bLogFile = new JToggleButton("Logging");
            bLogFile.addActionListener(this);
            p.add(bLogFile);
            tLogFile = new JTextField(15);
            p.add(tLogFile);
            bScanUsb = new JButton("Scan USB");
            bScanUsb.addActionListener(this);
            p.add(bScanUsb);
            bPoll = new JButton("Poll Unconnected");
            bPoll.addActionListener(this);
            p.add(bPoll);
            bPollAll = new JButton("Poll All");
            bPollAll.addActionListener(this);
            p.add(bPollAll);
            add(new JScrollPane(p), BorderLayout.NORTH);
            
            p = new JPanel(new GridLayout(5,5));
			JScrollPane jsp = new JScrollPane(p);
            add(jsp, BorderLayout.CENTER);
            
            seedSelect = new ButtonGroup();
            
            // Create a GUI for each node
            nodes = new NodeSquare[TESTBED_SIZE];
            for (int i=0; i<nodes.length; i++)
            {
                nodes[i] = new NodeSquare();
                nodes[i].setBorder(BorderFactory.createTitledBorder("Node "+(i)));