The socket variables last_remote_ip, last_remote_port, last_local_ip * store the source and destination of the last received UDP/IP - * header if this is a 'listening server socket', and they can be - * retrieved by the application that opened this socket. + * header. * *
UDP has no flow control, therefore for a UDP socket the write() * operation sends nbytes of data (virtual or an object) @@ -53,17 +49,19 @@ public ProtocolSession appsess; // we initalize these to -1 to help distinguish a fully configured - // socket from an "unconnected" listening server socket + // socket from public int local_ip = -1; public int remote_ip = -1; public int local_port = -1; public int remote_port = -1; + private int[] remote; // variables to save the source and dest variables of the last received // UDP/IP headers to pass them to the owner application if required public int last_remote_ip; public int last_local_ip; public int last_remote_port; + private int[] last_addr; // empty dataMessage for sending virtual data can be safely recycled // because it is not a payload of a UDP_Header @@ -91,19 +89,18 @@ * this socket after a UDP session is created via listen() or connect(). */ public void bind(int my_ip, int my_port){ + if (local_ip!=-1) { + System.err.println("Warning: rebinding udpSocket."); + return; + } local_ip = my_ip; local_port = my_port; + udpsess = udpsessMaster.openSession(this); } - /** after bind(), create a UDP listening session so that read() may be used. - * request_limit is not implemented. - */ + /** not supported for udp sockets */ public void listen(int request_limit) { - if(local_port == -1) { - System.err.println("udpSocket Fatal Error: bind() local port before listen()."); - System.exit(-1); - } - udpsess = udpsessMaster.openSession(this); + System.err.println("Warning: udpSocket does not implement listen()."); } /** sets remote IP and UDP port number and creates a UDP session so that @@ -111,9 +108,9 @@ */ public void connect(int dest_ip, int dest_port, Continuation caller) throws ProtocolException { - remote_ip = dest_ip; - remote_port = dest_port; - udpsess = udpsessMaster.openSession(this); + remote=new int[2]; + remote[0]=dest_ip; + remote[1]=dest_port; } /** not supported for udp sockets */ @@ -122,23 +119,7 @@ } public void write(Object[] obj, int nbytes, Continuation caller){ - if(nbytes > udpsessMaster.max_datagram_size) { - caller.failure(1); - return; - } - if (obj[0] == null) { - System.err.println("Warning: udpSocket.write() null object - send as virtual data."); - write(nbytes, caller); - return; - } - dataMessage dmsg = new dataMessage(obj[0], nbytes); - try { - if (udpsess.push(dmsg, null)) - caller.success(); - else - caller.failure(2); - } - catch (ProtocolException e) { System.err.println(e);} + sendto(obj, nbytes, null, caller); } /** not supported for udp sockets */ @@ -147,13 +128,37 @@ } public void write(int nbytes, Continuation caller){ + sendto(null, nbytes, null, caller); + } + + public void sendto(Object[] obj, int nbytes, int[] addr, Continuation caller) { + if (addr == null) { + if (remote == null) { + System.err.println("Warning: udpSocket.write() without connect()."); + caller.failure(3); + return; + } + addr = remote; + } + if (obj != null && obj[0] == null) { + System.err.println("Warning: udpSocket.write() null object - send as virtual data."); + obj = null; + } if(nbytes > udpsessMaster.max_datagram_size) { caller.failure(1); return; } - emptyMsg.size = nbytes; + dataMessage dmsg=null; + if (obj!=null) + dmsg = new dataMessage(obj[0], nbytes); + else { + emptyMsg.size = nbytes; + dmsg = emptyMsg; + } + remote_ip=addr[0]; + remote_port=addr[1]; try { - if (udpsess.push(emptyMsg, null)) + if (udpsess.push(dmsg, null)) caller.success(); else caller.failure(2); @@ -172,25 +177,24 @@ } public void read(Object[] obj, int nbytes, Continuation caller) { - if(readCont == null) { - readCont = caller; - readSize = nbytes; - dataObj = obj; - } else { - System.err.println("udpSocket: read request error - socket already reading"); - } + recvfrom(obj, nbytes, null, caller); } public void read(int nbytes, Continuation caller){ + recvfrom(null, nbytes, null, caller); + } + + public void recvfrom(Object[] obj, int nbytes, int[] addr, Continuation caller) { if(readCont == null) { readCont = caller; readSize = nbytes; - dataObj = null; + dataObj = obj; + last_addr = addr; } else { System.err.println("udpSocket: read request error - socket already reading"); } } - + public void close(Continuation caller)throws ProtocolException{ udpsessMaster.closeSession(udpsess); } @@ -202,15 +206,21 @@ if (readCont == null) return; - rcv_buf_len += msg.size; + //rcv_buf_len += msg.size; if((msg.data != null) && (dataObj != null)) dataObj[0] = msg.data; - if(readSize <= rcv_buf_len){ - rcv_buf_len -= readSize; + // Is this correct? --jop + //if(readSize <= rcv_buf_len){ + //rcv_buf_len -= readSize; + if (last_addr!=null) { + last_addr[0] = last_remote_ip; + last_addr[1] = last_remote_port; + } + last_addr = null; temp = readCont; readCont = null; temp.success(); - } + //} } } --- orig/src/SSF/OS/UDP/test/udpStreamServer.java +++ mod/src/SSF/OS/UDP/test/udpStreamServer.java @@ -58,7 +58,7 @@ /** size of data request object from matching client */ int request_size; - /** "listening" socket of this server */ + /** socket of this server */ udpSocket udpsocket; /** max number of clients of this server - default is unlimited */ @@ -95,31 +95,18 @@ int numDatagrams = 0; int numFailures = 0; udpSocket sd; - int srcIP; - int srcPort; - int destIP; - int destPort; + int[] dest; - public slaveServer(int docsize, int local_ip, - int local_port, int remote_ip, int remote_port) { + public slaveServer(int docsize, udpSocket sd, int[] remote) { owner = udpStreamServer.this; nbytesRemaining = docsize; nbytesRequest = docsize; - srcIP = local_ip; - srcPort = local_port; - destIP = remote_ip; - destPort = remote_port; + this.sd=sd; + dest=remote; dataSize = owner.datagram_size; } void start() { - try { - sd = (udpSocket)owner.socketms.socket(this, "udp"); - sd.bind(srcIP, srcPort); - sd.connect(destIP, destPort, null); // creates a UDP session - } catch (ProtocolException e) { - System.err.println(e); - } senddata(); } @@ -129,7 +116,7 @@ public void callback(){ if (nbytesRemaining > 0){ if((nbytesRemaining - dataSize) < 0) dataSize = nbytesRemaining; - sd.write(dataSize, new Continuation(){ + sd.sendto(null, dataSize, dest, new Continuation(){ public void success() { nbytesRemaining -= dataSize; if (debug) { @@ -200,17 +187,17 @@ /** preamble to slave server diagnostics */ void localInfo(String str){ System.out.println(owner.inGraph().now()/(double)SSF.Net.Net.seconds(1.0) - + " udpServer " + IP_s.IPtoString(srcIP) - + ":" + srcPort + " " + str); + + " udpServer " + IP_s.IPtoString(localIP) + + ":" + wellKnownPort + " " + str); } /** preamble to end2end UDP session diagnostics */ void sessionInfo(String str){ System.out.println(owner.inGraph().now()/(double)SSF.Net.Net.seconds(1.0) - + " udpServer " + IP_s.IPtoString(srcIP) - + ":" + srcPort - + " clnt " + IP_s.IPtoString(destIP) - + ":" + destPort + " " + str); + + " udpServer " + IP_s.IPtoString(localIP) + + ":" + wellKnownPort + + " clnt " + IP_s.IPtoString(dest[0]) + + ":" + dest[1] + " " + str); } public boolean push(ProtocolMessage message, ProtocolSession fromSession) @@ -289,14 +276,14 @@ }catch (ProtocolException e) { System.err.println("udpServer: " + e); } - udpsocket.bind(-1, wellKnownPort); - udpsocket.listen(client_limit); // creates a UDP session + udpsocket.bind(localIP, wellKnownPort); obj = new Object[1]; serv(); } public void serv() { - udpsocket.read(obj, request_size, new Continuation(){ + final int[] addr=new int[2]; + udpsocket.recvfrom(obj, request_size, addr, new Continuation(){ public void success() { if((obj == null) || (obj[0] == null)) { serverInfo("received unknown request"); @@ -314,11 +301,7 @@ // a slave server reuses the master server's port number, // and uses the local IP, remote port and remote IP that were // contained in the service request UDP header - slaveServer srv = new slaveServer(doc_size, - udpsocket.last_local_ip, - wellKnownPort, - udpsocket.last_remote_ip, - udpsocket.last_remote_port ); + slaveServer srv = new slaveServer(doc_size, udpsocket, addr); srv.start(); } } --- orig/src/SSF/OS/UDP/test/udpStreamSession.java +++ mod/src/SSF/OS/UDP/test/udpStreamSession.java @@ -116,7 +116,7 @@ } public void failure(int errno) { if (showDebug) { - str = "request " + " to " + srvNHI + " FAILED"; + str = "request " + " to " + srvNHI + " FAILED "+errno; clientDebug(str); } } --- orig/src/SSF/OS/TCP/tcpSocket.java +++ mod/src/SSF/OS/TCP/tcpSocket.java @@ -489,6 +489,10 @@ myTcpSession.write(emptyMsg, caller); } + public void sendto(Object[] buf, int nbytes, int[] remote, Continuation caller) { + write(buf, nbytes, caller); + } + /** not supported by tcp sockets */ public void read(byte[] buf, int nbytes, Continuation caller) { socketInfo("Warning: tcpSocket does not implement byte read()."); @@ -530,4 +534,8 @@ appCallWaiting = true; myTcpSession.read(null, nbytes, caller); } + + public void recvfrom(Object[] obj, int nbytes, int[] remote, Continuation caller) { + read(obj, nbytes, caller); + } }