Next step: Java NIO for the Java client code.
#include <arpa/inet.h>
#include <strings.h>
#include <pthread.h>
-#include <semaphore.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/wait.h>
int returnValue = -1; /*Return value from this function can be caught by upper*/
/*layers, NOK by default*/
+
+ // TODO: stop using shared memory. There must be another way to do this.
+ // See OpenJDK Fork Process, it uses a file descriptor.
/*
* Allocate shared memory for the return status code from the process which is
int pid; /*Child or parent PID.*/
int out[2], err[2]; /*Store pipes file descriptors. Write ends attached to the stdout*/
/*and stderr streams.*/
- char buf[2000]; /*Read data buffer. allignment(int) * 500.*/
- char string[3000];
+ u_char buf[2000]; /*Read data buffer. allignment(int) * 500.*/
struct pollfd polls[2]; /*pipes attached to the stdout and stderr streams.*/
int n; /*characters number from stdout and stderr*/
- struct tcpforkhdr header;
+ struct tcpforkhdr *header = (struct tcpforkhdr *)buf;
int childreturnstatus;
int returnValue = 0; /*return value from this function can be caught by upper layers,*/
/*OK by default*/
polls[1].fd=err[0];
polls[0].events = polls[1].events = POLLIN;
- bzero(string, sizeof(string));
- /*TODO: stop using XML. Next improvements: my own client/server protocol*/
- sprintf(string,"<?xml version=\"1.0\"?><streams>");
-
- if (TEMP_FAILURE_RETRY(send(socket,string,strlen(string),0)) < 0)
- syslog (LOG_INFO, "error while sending xml header: %m");
-
-
for (;;) {
if(poll(polls, 2, 100)) {
if(polls[0].revents && POLLIN) {
bzero(buf,2000);
- bzero(string, sizeof(string));
- n=TEMP_FAILURE_RETRY(read(out[0], &buf[1], 1999));
- bzero(&header, sizeof(header));
+ n=TEMP_FAILURE_RETRY(read(out[0], &buf[sizeof(struct tcpforkhdr)], 2000-sizeof(struct tcpforkhdr)));
//To network order, indeed it is the order used by Java (BIG ENDIAN). Anyway I am
//swapping the bytes because it is required if you want to write portable code and
//ENDIANNESS indepedent.
- header.type = htonl(1);
- header.type = htonl(n);
+ header->type = htonl(1);
+ header->length = htonl(n);
//PACKING THE STRUCT OR SERIALIZING?
//serializing-> sends the struct one data member at time (using for example writev?) I send
//one field of my struct at time.
//and the client application must know what it has to do with them.
//TODO: my own protocol to make the client independent of the ENDIANNESS and character set used
//by the machine running this server. See comments in the TCPForkDaemon.java code about this.
- sprintf(string,"<out><![CDATA[%s]]></out>", buf);
- if (TEMP_FAILURE_RETRY(send(socket,string,strlen(string),0)) < 0)
+ if (TEMP_FAILURE_RETRY(send(socket, buf, n+sizeof(struct tcpforkhdr), 0)) < 0)
syslog (LOG_INFO, "error while sending stdout: %m");
}
if(polls[1].revents && POLLIN) {
bzero(buf,2000);
- bzero(string, sizeof(string));
- n=TEMP_FAILURE_RETRY(read(err[0],buf,1990));
- sprintf(string,"<error><![CDATA[%s]]></error>", buf);
- if (TEMP_FAILURE_RETRY(send(socket,string,strlen(string),0)) < 0)
- syslog (LOG_INFO, "error while sending stderr: %m");
+ n=TEMP_FAILURE_RETRY(read(err[0], &buf[sizeof(struct tcpforkhdr)], 2000-sizeof(struct tcpforkhdr)));
+ header->type = htonl(2);
+ header->length = htonl(n);
+ if (TEMP_FAILURE_RETRY(send(socket, buf, n+sizeof(struct tcpforkhdr), 0)) < 0)
+ syslog (LOG_INFO, "error while sending stdout: %m");
}
if(!(polls[0].revents && POLLIN) && !(polls[1].revents && POLLIN)) {
syslog (LOG_ERR, "error while waiting for killed child process: %m");
}
- /*In Java the client will get a XMLParser Exception.*/
+ /*In the Java code, the client will get an error as the return status from the exec method.*/
goto err;
}
}
else {
/*When timeout*/
if(TEMP_FAILURE_RETRY(waitpid(pid, &childreturnstatus, WNOHANG))) {
+ //TODO: treating errors in waitpid function
/*Child is dead, we can finish the connection*/
/*First of all, we check the exit status of our child process*/
/*In case of error send an error status to the remote calling process*/
}
}
}
- /*Reaching this code when child finished or if error while polling pipes*/
- bzero(string, sizeof(string));
- sprintf(string,"<ret><![CDATA[%d]]></ret></streams>", (*returnstatus));
- if (TEMP_FAILURE_RETRY(send(socket,string,strlen(string),0)) < 0)
+ /*Stuff just done by the parent process. The child process ends with exit*/
+ /*Reaching this code when child ends*/
+ bzero(buf,2000);
+ header->type = htonl(3);
+ header->length = htonl((*returnstatus));
+ if (TEMP_FAILURE_RETRY(send(socket, buf, sizeof(struct tcpforkhdr), 0)) < 0)
syslog (LOG_INFO, "error while sending return status: %m");
- /*Stuff just done by the Parent process. The child process ends with exit*/
end:
closeSafely (out[0]);
struct tcpforkhdr{
//In this way, there are not issues related to ENDIANNESS or padding
//but we are wasting bytes for the type field...
- uint32_t type; /*Data alignment: 4-byte aligned*/
- uint32_t length; /*Data alignment: 4-byte aligned*/
+ uint32_t type; /*Data alignment: 4-byte aligned. For Java, we must send the integer using BIG ENDIAN*/
+ uint32_t length; /*Data alignment: 4-byte aligned. For Java, we must send the integer using BIG ENDIAN*/
//We use fixed width integer types from C99.
};
import java.io.InputStream;
import java.io.PrintStream;
import java.net.UnknownHostException;
-import javax.xml.parsers.ParserConfigurationException;
import org.apache.log4j.Logger;
-import org.xml.sax.SAXException;
/**
*
*
* @return return code.
*/
- public static int exec(final String command) throws IOException, InterruptedException {
+ public static int exec(final String command) throws IOException {
return exec(command, null, null);
}
*
* @return return code.
*/
- public static int exec(final String command, final PrintStream standarOutPut) throws IOException, InterruptedException {
+ public static int exec(final String command, final PrintStream standarOutPut) throws IOException {
return exec(command, standarOutPut, null);
}
*
* @return return code from the executed system command.
*/
- public static int exec(final String command, final PrintStream standarOutPut, final PrintStream errorOutPut) throws IOException, InterruptedException {
+ public static int exec(final String command, final PrintStream standarOutPut, final PrintStream errorOutPut) throws IOException {
return exec(command, standarOutPut, errorOutPut, DEFAULT_HOST, DEFAULT_PORT);
}
* @param command system command to be executed.
* @param aLogger send the information to log.
*/
- public static int exec(final String command, final Logger aLogger) throws IOException, InterruptedException {
+ public static int exec(final String command, final Logger aLogger) throws IOException {
//calling private method to handle logger input/ouput in a common method
return execHandlingLogger(command, aLogger, DEFAULT_HOST, DEFAULT_PORT);
* @throws IOException
* @throws InterruptedException
*/
- public static int exec(final String[] commandAndArguments, final Logger aLogger) throws IOException, InterruptedException {
+ public static int exec(final String[] commandAndArguments, final Logger aLogger) throws IOException {
String wholeCommand="";
for(String argument : commandAndArguments) {
* @throws IOException
*/
public static int exec(final String command, final PrintStream standarOutPut,
- final PrintStream errorOutPut, final String host, final int port)
- throws IOException, InterruptedException {
+ final PrintStream errorOutPut, final String host, final int port) throws IOException {
int exitStatus = LauncherProcesses.STATUS_ERR;
- XmlForkParser forkParser = null;
TCPForkDaemon process = null;
- try {
- forkParser = new XmlForkParser();
- process = new TCPForkDaemon(forkParser, host, port);
- exitStatus = process.exec(command);
- } catch (ParserConfigurationException e) {
- // This is not a crazy thing, we are trying to insert this new method without
- // breaking the old methods which did not throw SAXException or ParserConfigurationException
- // Do not blame me.
- throw new IOException(e);
- } catch (SAXException e) {
- // This is not a crazy thing, we are trying to insert this new method without
- // breaking the old methods which did not throw SAXException or ParserConfigurationException
- // Do not blame me.
- throw new IOException(e);
- }
-
+
+ process = new TCPForkDaemon(host, port);
+ exitStatus = process.exec(command);
if ((standarOutPut != null) && (process.getStdout() != null)){
* @throws InterruptedException
*/
private static int execHandlingLogger(final String command, final Logger aLogger,
- final String host, int port) throws IOException, InterruptedException {
+ final String host, int port) throws IOException {
int exitStatus = LauncherProcesses.STATUS_ERR;
- XmlForkParser forkParser = null;
TCPForkDaemon process = null;
- try {
- forkParser = new XmlForkParser();
- process = new TCPForkDaemon(forkParser, host, port);
- exitStatus = process.exec(command);
- } catch (ParserConfigurationException e) {
- // This is not a crazy thing, we are trying to insert this new method without
- // breaking the old methods which did not throw SAXException or ParserConfigurationException
- // Do not blame me.
- throw new IOException(e);
- } catch (SAXException e) {
- // This is not a crazy thing, we are trying to insert this new method without
- // breaking the old methods which did not throw SAXException or ParserConfigurationException
- // Do not blame me.
- throw new IOException(e);
- }
-
+
+ process = new TCPForkDaemon(host, port);
+ exitStatus = process.exec(command);
if (process.getStdout() != null) {
* @throws IOException
* @throws InterruptedException
*/
- public static InputStream execStream (final String [] command, final Logger aLogger)
- throws IOException, InterruptedException {
+ public static InputStream execStream (final String [] command, final Logger aLogger) throws IOException {
int exitStatus = LauncherProcesses.STATUS_ERR;
InputStream stdInput = null;
- XmlForkParser forkParser = null;
TCPForkDaemon process = null;
String wholeCommand="";
wholeCommand = wholeCommand + " " + argument;
}
- try {
- forkParser = new XmlForkParser();
- process = new TCPForkDaemon(forkParser, DEFAULT_HOST, DEFAULT_PORT);
- exitStatus = process.exec(wholeCommand);
- } catch (ParserConfigurationException e) {
- throw new IOException(e);
- } catch (SAXException e) {
- throw new IOException(e);
- }
+
+ process = new TCPForkDaemon(DEFAULT_HOST, DEFAULT_PORT);
+ exitStatus = process.exec(wholeCommand);
if(exitStatus == 0) {
+ //It must be used the remote charset.
+ //TODO: my own protocol
stdInput = new ByteArrayInputStream(process.getStdout().getBytes("UTF-8"));
}
else {
* @throws IOException
* @throws InterruptedException
*/
- public static int execInLocation (final String command, final String location) throws IOException, InterruptedException {
+ public static int execInLocation (final String command, final String location) throws IOException {
int exitStatus = LauncherProcesses.STATUS_ERR;
final String wholeCommand = "cd " + location + " && " + command;
final ByteArrayOutputStream baos = new ByteArrayOutputStream();
final PrintStream stdout = new PrintStream(baos);
- final String command = "ls -lah ~/Desktop; ls -lah * bbbb aaa";
+ final String command = "sleep 10; ls -lah";
final ByteArrayOutputStream baos2 = new ByteArrayOutputStream();
final PrintStream stderr = new PrintStream(baos2);
int result;
package de.fork.java;
-import java.io.DataOutputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.EOFException;
import java.io.IOException;
-import java.io.PrintWriter;
import java.net.InetAddress;
import java.net.Socket;
import java.net.UnknownHostException;
-import org.xml.sax.InputSource;
-import org.xml.sax.SAXException;
+
/**
+ * TODO: JAVA NIO (for example ByteBuffer, Socket and Selects)
* <p>
* With this class we can run processes using the intended daemon which is
* waiting for TCP connections in a specified port.
* </p>
*/
public class TCPForkDaemon {
- private final XmlForkParser parser;
private final String host;
private final int port;
+ private String streamStdout;
+ private String streamStderr;
+ //Error by default.
+ private int results = -1;
/**
* supported. See <a href="Inet6Address.html#scoped">here</a> for a description of IPv6
* scoped addresses.
* </p>
- * @param parser instance implemeting {@link XmlForkParser} which knows about what
- * codification uses the daemon to send us the results of the command sent to
- * by the remote daemon by the {@link TCPForkDaemon.#exec(String)} method.
* @param host the specified host.
* @param port the TCP port where the daemon accepts connections.
*
*/
- public TCPForkDaemon (final XmlForkParser parser, final String host, final int port) {
- this.parser = parser;
+ public TCPForkDaemon (final String host, final int port) {
this.host = host;
this.port = port;
}
* @return the executed command's return code.
* @throws IOException
* @throws UnknownHostException
- * @throws SAXException
* @throws SecurityException if a security manager exists
*/
- public int exec(final String command) throws UnknownHostException, IOException, SAXException {
- PrintWriter out = null;
+ public int exec(final String command) throws UnknownHostException, IOException {
Socket socket = null;
+ ByteArrayOutputStream lengthAndCommand = null;
+ DataInputStream receiveData = null;
+ ByteArrayOutputStream stdout = new ByteArrayOutputStream();
+ ByteArrayOutputStream stderr = new ByteArrayOutputStream();
/******************************************************************************************/
/* Just over 1 TCP connection */
/* JAVA CLIENT: <---------- CLOSE CONNECTION ------- :SERVER */
/* */
/******************************************************************************************/
-
-
- //SocketChannel socketNIO = new SocketChannelImpl();
socket = new Socket(InetAddress.getByName(host), port);
try {
//In this way we do not need to know about the remote encoding and everything could run automatically.
byte [] commandEncoded = command.getBytes("UTF-8");
- DataOutputStream sendData = new DataOutputStream(socket.getOutputStream());
// 1. COMMAND_LENGTH
- // This method sends the data like bytes using 4 system calls!!! Welcome to Java World...
- sendData.writeInt(commandEncoded.length);
+ lengthAndCommand = new ByteArrayOutputStream (commandEncoded.length + 4);
+ int v = commandEncoded.length;
+ lengthAndCommand.write((v >>> 24) & 0xFF);
+ lengthAndCommand.write((v >>> 16) & 0xFF);
+ lengthAndCommand.write((v >>> 8) & 0xFF);
+ lengthAndCommand.write((v >>> 0) & 0xFF);
// 2. COMMAND
- sendData.write(commandEncoded);
+ lengthAndCommand.write(commandEncoded);
+
+ // Sending length and command in the same chunk.
+ lengthAndCommand.writeTo(socket.getOutputStream());
- //DataInputStream receiveData = new DataInputStream(socket.getInputStream());
- //receiveData.read
// 3. RESULTS
// TODO: When the network infrastructure (between client and server) fails in this point
// Impossible to use a timeout, because we do not know how much time is going to long the command :/
// the only way to fix this issue in Java is sending ping messages (Could we fix it using custom settings in the OS
// of the client and server machines? for example in Linux see /proc/sys/net/ipv4/)
- InputSource inputSource = new InputSource(socket.getInputStream());
//It must be used the remote locale character set encoding. If using for example multi-byte system (like UTF-16)
//you must know about the ENDIANNESS for the remote machine.
//TODO: my own protocol:
//C server knows about its ENDIANNESS and character set so it can encode to TCPForkDaemon encode (which could be UTF-8) ->
//sends response to Java client and here we know the character set encoding is the defined for TCPForkDaemon protocol
- //(for example UTF-8) In this way we do not need to know about the remote encoding and everything could run automatically.
- inputSource.setEncoding("UTF-8");
- parser.setStream(socket.getInputStream());
-
+ //(for example UTF-8) In this way we do not need to know about the remote encoding and everything could run automatically.
+ receiveData = new DataInputStream(socket.getInputStream());
+ while (true) {
+ int type = receiveData.readInt();
+ int length = receiveData.readInt();
+ switch (type) {
+ case 0:
+ //STDIN not implemented
+ break;
+ case 1:
+ //STDOUT
+ byte [] dataStdout = new byte[length];
+ receiveData.readFully(dataStdout, 0, length);
+ stdout.write(dataStdout, 0, length);
+ break;
+ case 2:
+ //STDERR
+ byte [] dataStderr = new byte[length];
+ receiveData.readFully(dataStderr, 0, length);
+ stderr.write(dataStderr, 0, length);
+ break;
+ case 3:
+ //RESULTS
+ results = length;
+ break;
+ default:
+ throw new IOException("Unrecognized type.");
+
+ }
+ }
+ } catch (EOFException e) {
// 4. SERVER CLOSED CONNECTION
+ //TODO: Java NIO with select, and stop using this crappy code.
}
finally {
- if (out != null) {
- out.close();
+ if (lengthAndCommand != null) {
+ lengthAndCommand.close();
+ }
+ if (receiveData != null) {
+ //Closes input stream and (of course) the socket
+ receiveData.close();
+ }
+ if (!socket.isClosed()) {
+ //In case of not closing the socket with receiveData.
+ socket.close();
}
- socket.close();
+
+ //The remote charset.
+ //TODO: my own protocol as above specified.
+ this.streamStderr = new String(stderr.toByteArray(), "UTF-8");
+ this.streamStdout = new String(stdout.toByteArray(), "UTF-8");
}
- //If everything went alright we should be able to retrieve the return
+ //If everything went all right we should be able to retrieve the return
//status of the remotely executed command.
- return parser.getReturnValue();
+ return this.results;
+
}
* @return the stdout stream
*/
public String getStdout() {
- return parser.getStdout();
+ return this.streamStdout;
}
* @return the stderr stream
*/
public String getStderr() {
- return parser.getStderr();
+ return this.streamStderr;
}
}
+++ /dev/null
-package de.fork.java;
-
-import java.io.IOException;
-import java.io.InputStream;
-import javax.xml.parsers.ParserConfigurationException;
-import javax.xml.parsers.SAXParser;
-import javax.xml.parsers.SAXParserFactory;
-import org.apache.log4j.Logger;
-import org.xml.sax.Attributes;
-import org.xml.sax.SAXException;
-import org.xml.sax.SAXParseException;
-import org.xml.sax.ext.DefaultHandler2;
-
-/**
- * <p>
- * Class intended to parse the XML stream received from the daemon which is
- * waiting to run commands. These commands are sent by the method
- * {@link de.fork.java.TCPForkDaemon#exec(String, String, int)}
- * </p>
- * <p>
- * After processing one command the daemon sends a XML stream with the stderr,
- * stdout and return status of that command. With this class we extract those values
- * and we can retrieve them using the methods {@link #getStderr() }, {@link #getStdout()}
- * and {@link #getReturnValue()}
- * </p>
- * <p>
- * <pre>
- * <b>Example, stream received from daemon:</b>
- * {@code
- * <?xml version="1.0"?><salida><error><![CDATA[ls: no se puede acceder a bbb: No existe el fichero o el directorio
- * ls: no se puede acceder a aaa: No existe el fichero o el directorio
- * ls: no se puede acceder a dddd: No existe el fichero o el directorio
- * ]]></error><ret><![CDATA[2]]></ret></salida>
- * }
- * </pre>
- * </p>
- * <p>
- * Instances of this class are mutable. To use them concurrently, clients must surround each
- * method invocation (or invocation sequence) with external synchronization of the clients choosing.
- * </p>
- */
-public class XmlForkParser extends DefaultHandler2 {
- private static final Logger logger = Logger.getLogger(XmlForkParser.class);
- private StringBuffer accumulator = new StringBuffer();
- private StringBuilder stderr = new StringBuilder();
- private String lastStderr;
- private StringBuilder stdout = new StringBuilder();
- private String lastStdout;
- private String returnCode = new String();
- final SAXParserFactory spf = SAXParserFactory.newInstance();
- private final SAXParser saxParser;
-
- public XmlForkParser() throws ParserConfigurationException, SAXException {
- saxParser = spf.newSAXParser();
- }
-
- public void setStream(InputStream stream) throws SAXException, IOException {
- saxParser.parse(stream, this);
- }
-
- /**
- * <p>
- * The daemon sends a XML stream, we parse that stream and the results are
- * stored in the instace fields {@link #stderr}, {@link #stdout} and {@link returnCode}
- * </p>
- * <p>
- * Later we can retrieve the results with {@link #getStderr()}, {@link #getStdout()} and
- * {@link #getReturnValue()}
- * </p>
- */
- @Override
- public void endElement (final String uri, final String localName, final String qName) {
- if (qName.equals("error")) {
- // After </error>, we've got the stderror
- stderr = stderr.append(accumulator.toString());
- } else if (qName.equals("out")) {
- // After </out>, we've got the stdout
- stdout = stdout.append(accumulator.toString());
- } else if (qName.equals("ret")) {
- returnCode = returnCode + accumulator.toString();
- }
- }
-
- /**
- * <p>
- * This method removes the <code>\n<code> characters at the end of the stdout
- * or stderr stream.
- * </p>
- *
- * @throws SAXException If any SAX errors occur during processing.
- */
- @Override
- public void endDocument () throws SAXException
- {
- if (stderr.length() != 0) {
- lastStderr = stderr.toString().replaceFirst("\\\n$", "");
- }
- if (stdout.length() != 0) {
- lastStdout = stdout.toString().replaceFirst("\\\n$", "");
- }
- }
-
- /**
- * Retrieve the standard error.
- * When there is nothing from the standard error this method returns null.
- *
- * <pre>
- * <b>Example, stream received from daemon:</b>
- * {@code
- * <?xml version="1.0"?><salida><error><![CDATA[ls: no se puede acceder a bbb: No existe el fichero o el directorio
- * ls: no se puede acceder a aaa: No existe el fichero o el directorio
- * ls: no se puede acceder a dddd: No existe el fichero o el directorio
- * ]]></error><ret><![CDATA[2]]></ret></salida>
- * }
- * </pre>
- * </p>
- * <p>
- * <pre>
- * <b>From that example with this method we are going to obtain this return parameter:</b>
- * {@code
- * ls: no se puede acceder a bbb: No existe el fichero o el directorio
- * ls: no se puede acceder a aaa: No existe el fichero o el directorio
- * ls: no se puede acceder a dddd: No existe el fichero o el directorio
- * }
- * </pre>
- *
- * @return stderr
- */
- public String getStderr() {
- return lastStderr;
-
- }
-
-
- /**
- * Retrieve the standard output.
- * When there is nothing from the standard output this method returns null.
- *
- * @see {@link XmlForkParser#getStderr()}
- * @return stdout
- */
- public String getStdout() {
- return lastStdout;
- }
-
-
- /**
- * Retrieve the return code from the executed command.
- *
- * @return return status, usually <code>0<code> means the command went OK.
- */
- public int getReturnValue() {
- return new Integer(returnCode).intValue();
- }
-
-
- @Override
- public void startElement (final String uri, final String localName,
- final String qName, final Attributes attributes) {
- accumulator.setLength(0);
- }
-
-
- @Override
- public void characters(final char[] buffer, final int start, final int length) {
- accumulator.append(buffer, start, length);
- }
-
-
- @Override
- public void warning(final SAXParseException exception) {
- logger.error("WARNING line:" + exception.getLineNumber(), exception);
- }
-
-
- @Override
- public void error(final SAXParseException exception) {
- logger.error("ERROR line:" + exception.getLineNumber(), exception);
- }
-
-
- @Override
- public void fatalError(final SAXParseException exception) throws SAXException {
- logger.error("FATAL ERROR line:" + exception.getLineNumber(), exception);
- throw (exception);
- }
-}
projects / JavaTCPFork / .git / commitdiff