uima-ducc-sm/src/main/java/org/apache/uima/ducc/sm/ServicePingMain.java - uima-ducc - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
 */
 package org.apache.uima.ducc.sm;

 import java.io.IOException;
 import java.io.InputStream;
 import java.io.ObjectOutputStream;
 import java.io.OutputStream;
 import java.net.Socket;
 import java.net.UnknownHostException;

 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.CommandLineParser;
 import org.apache.commons.cli.OptionBuilder;
 import org.apache.commons.cli.Options;
 import org.apache.commons.cli.ParseException;
 import org.apache.commons.cli.PosixParser;
 import org.apache.uima.ducc.cli.AServicePing;
 import org.apache.uima.ducc.cli.ServiceStatistics;
 import org.apache.uima.ducc.common.IServiceStatistics;


 /**
  * If an external pinger is specified for a service, this method instantiates and executes
  * the pinger.
  *
  * The pinger must extend org.apache.uima.ducc.sm.cli.ServicePing and implement the ping() method.
  *
  */

 public class ServicePingMain
     implements SmConstants
 {

     /**
 	 *
 	 */
 	private static final long serialVersionUID = 1L;
 	boolean debug = false;
     int error_max = 10;
     int error_count = 0;

     public ServicePingMain()
     {

     }

 	@SuppressWarnings("static-access")
 	private void addOptions(Options options)
     {
         //
         // Verbs here
         //
 		options.addOption(OptionBuilder
                           .withLongOpt    (ServicePing.Class.decode())
                           .withDescription(ServicePing.Class.description())
                           .withArgName    (ServicePing.Class.argname())
                           .hasOptionalArg ()
                           .create         ()
                           );

 		options.addOption(OptionBuilder
                           .withLongOpt    (ServicePing.Endpoint.decode())
                           .withDescription(ServicePing.Endpoint.description())
                           .withArgName    (ServicePing.Endpoint.argname())
                           .hasArg         (true)
                           .create         ()
                           );

 		options.addOption(OptionBuilder
                           .withLongOpt    (ServicePing.Port.decode())
                           .withDescription(ServicePing.Port.description())
                           .withArgName    (ServicePing.Port.argname())
                           .hasArg         (true)
                           .create         ()
                           );

 		options.addOption(OptionBuilder
                           .withLongOpt    (ServicePing.Port.decode())
                           .withDescription(ServicePing.Port.description())
                           .withArgName    (ServicePing.Port.argname())
                           .hasArg         (true)
                           .create         ()
                           );

     }

     static void appendStackTrace(StringBuffer s, Throwable t)
     {
     	s.append("\nAt:\n");
         StackTraceElement[] stacktrace = t.getStackTrace();
         for ( StackTraceElement ste : stacktrace ) {
             s.append("\t");
             s.append(ste.toString());
             s.append("\n");
         }
     }


     public static void print(Object ... args)
     {
     	StringBuffer s = new StringBuffer();
         for ( Object a : args ) {
             if ( a == null ) a = "<null>"; // avoid null pointers

             s.append(" ");
             if ( a instanceof Throwable ) {
             	Throwable t = (Throwable ) a;
                 s.append(t.toString());
                 s.append("\n");
                 appendStackTrace(s, t);
             } else {
                 s.append(a.toString());
             }
         }
         System.err.println(s.toString());
     }

     //
     // resolve the customMeta string inta a class if we can
     //
     AServicePing resolve(String cl, String args, String ep)
     {
     	print("ServicePingMain.resolve:", cl, "ep", ep);
     	AServicePing pinger = null;
 		try {
 			@SuppressWarnings("rawtypes")
 			Class cls = Class.forName(cl);
 			pinger = (AServicePing) cls.newInstance();
 			pinger.init(args, ep);
 		} catch (Exception e) {
             //print(e);         // To the logs
             e.printStackTrace();
 		}
         return pinger;
     }

     void handleError(Throwable t)
     {
         t.printStackTrace();
         if ( ++error_count >= error_max ) {
             System.out.println("Exceeded error count. Exiting.");
             System.exit(1);
         }
     }

     //
     // 1. Instantiate the pinger if possible.
     // 2. Read ducc.proeprties to find the ping interval
     // 3. Start pinging and wriging results to stdout
     //
     // The ServiceManager must start this process as the user.  It monitors stdout for success
     // or failute of the ping and reacts accordingly.
     //
     protected void start(String[] args)
     {


         Options options = new Options();
         addOptions(options);

         CommandLineParser parser = new PosixParser();
         CommandLine commandLine = null;
         IServiceStatistics default_statistics = new ServiceStatistics(false, false, "<N/A>");

 		try {
 			commandLine = parser.parse(options, args);
 		} catch (ParseException e) {
             print("Cannot parse command line:", e);
             return;
 		}

         String arguments = commandLine.getOptionValue(ServicePing.Arguments.decode());
         String pingClass = commandLine.getOptionValue(ServicePing.Class.decode());
         String endpoint  = commandLine.getOptionValue(ServicePing.Endpoint.decode());
         String port      = commandLine.getOptionValue(ServicePing.Port.decode());

         Socket sock = null;
 		try {
 			sock = new Socket("localhost", Integer.parseInt(port));
 		} catch (NumberFormatException e2) {
 			e2.printStackTrace();
 			return;
 		} catch (UnknownHostException e2) {
 			e2.printStackTrace();
 			return;
 		} catch (IOException e2) {
 			e2.printStackTrace();
 			return;
 		}
         print ("ServicePingMain listens on port", sock.getLocalPort());
         InputStream sock_in = null;
 		OutputStream sock_out = null;
 		try {
 			sock_in = sock.getInputStream();
 			sock_out = sock.getOutputStream();
 		} catch (IOException e2) {
 			e2.printStackTrace();
 			return;
 		}

         ObjectOutputStream oos;
 		try {
 			oos = new ObjectOutputStream(sock_out);
 			oos.flush();
 		} catch (IOException e1) {
 			e1.printStackTrace();
 			return;
 		}

         AServicePing custom = resolve(pingClass, arguments, endpoint);
         if ( custom == null ) {
             print("bad_pinger:", pingClass, endpoint);
             return;
         }

         while ( true ) {
         	if ( debug ) print("ServicePingMeta starts ping.");

             byte[] cmd = new byte[1];
             cmd[0] = 0;
             int eof = 0;
 			try {
 				eof = sock_in.read(cmd);
 			} catch (IOException e) {
                 handleError(e);
 			}
             if ( debug ) print("Read cmd", new String(cmd), "eof", eof);

             if ( eof == -1 ) {
                 print("EOF on input pipe.  Exiting");
                 custom.stop();
                 return;
             }

             try {
 				if ( cmd[0] == 'P' ) {
                     IServiceStatistics ss = custom.getStatistics();
                     if ( ss == null ) {
                         ss = default_statistics;
                     }
                     // print("Is alive: " + ss.isAlive());
                     oos.writeObject(ss);
                     oos.flush();

                     // The ObjectOutputStream will cache instances and if all you do is change a
                     // field or two in the object, it won't be detected and the stale object will be
                     // sent.  So you have to reset() the stream, (or use a new object, or use
                     // clone() here also if you want, but this is simplest and safest since we have
                     // no control over what the external pinger gives us.
                     oos.reset();
 				} else if ( cmd[0] == 'Q' ) {
 				    custom.stop();
 				    return;
 				} else {
 				    System.err.println("Invalid command recieved: " +  Byte.toString(cmd[0]));
 				}
 			} catch (Throwable e) {
                 handleError(e);
 			}
         }
     }

     public static void main(String[] args)
     {
         ServicePingMain wrapper = new ServicePingMain();
         wrapper.start(args);
     }

 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/
	package org.apache.uima.ducc.sm;

	import java.io.IOException;
	import java.io.InputStream;
	import java.io.ObjectOutputStream;
	import java.io.OutputStream;
	import java.net.Socket;
	import java.net.UnknownHostException;

	import org.apache.commons.cli.CommandLine;
	import org.apache.commons.cli.CommandLineParser;
	import org.apache.commons.cli.OptionBuilder;
	import org.apache.commons.cli.Options;
	import org.apache.commons.cli.ParseException;
	import org.apache.commons.cli.PosixParser;
	import org.apache.uima.ducc.cli.AServicePing;
	import org.apache.uima.ducc.cli.ServiceStatistics;
	import org.apache.uima.ducc.common.IServiceStatistics;


	/**
	* If an external pinger is specified for a service, this method instantiates and executes
	* the pinger.
	*
	* The pinger must extend org.apache.uima.ducc.sm.cli.ServicePing and implement the ping() method.
	*
	*/

	public class ServicePingMain
	implements SmConstants
	{

	/**
	*
	*/
	private static final long serialVersionUID = 1L;
	boolean debug = false;
	int error_max = 10;
	int error_count = 0;

	public ServicePingMain()
	{

	}

	@SuppressWarnings("static-access")
	private void addOptions(Options options)
	{
	//
	// Verbs here
	//
	options.addOption(OptionBuilder
	.withLongOpt (ServicePing.Class.decode())
	.withDescription(ServicePing.Class.description())
	.withArgName (ServicePing.Class.argname())
	.hasOptionalArg ()
	.create ()
	);

	options.addOption(OptionBuilder
	.withLongOpt (ServicePing.Endpoint.decode())
	.withDescription(ServicePing.Endpoint.description())
	.withArgName (ServicePing.Endpoint.argname())
	.hasArg (true)
	.create ()
	);

	options.addOption(OptionBuilder
	.withLongOpt (ServicePing.Port.decode())
	.withDescription(ServicePing.Port.description())
	.withArgName (ServicePing.Port.argname())
	.hasArg (true)
	.create ()
	);

	options.addOption(OptionBuilder
	.withLongOpt (ServicePing.Port.decode())
	.withDescription(ServicePing.Port.description())
	.withArgName (ServicePing.Port.argname())
	.hasArg (true)
	.create ()
	);

	}

	static void appendStackTrace(StringBuffer s, Throwable t)
	{
	s.append("\nAt:\n");
	StackTraceElement[] stacktrace = t.getStackTrace();
	for ( StackTraceElement ste : stacktrace ) {
	s.append("\t");
	s.append(ste.toString());
	s.append("\n");
	}
	}


	public static void print(Object ... args)
	{
	StringBuffer s = new StringBuffer();
	for ( Object a : args ) {
	if ( a == null ) a = "<null>"; // avoid null pointers

	s.append(" ");
	if ( a instanceof Throwable ) {
	Throwable t = (Throwable ) a;
	s.append(t.toString());
	s.append("\n");
	appendStackTrace(s, t);
	} else {
	s.append(a.toString());
	}
	}
	System.err.println(s.toString());
	}

	//
	// resolve the customMeta string inta a class if we can
	//
	AServicePing resolve(String cl, String args, String ep)
	{
	print("ServicePingMain.resolve:", cl, "ep", ep);
	AServicePing pinger = null;
	try {
	@SuppressWarnings("rawtypes")
	Class cls = Class.forName(cl);
	pinger = (AServicePing) cls.newInstance();
	pinger.init(args, ep);
	} catch (Exception e) {
	//print(e); // To the logs
	e.printStackTrace();
	}
	return pinger;
	}

	void handleError(Throwable t)
	{
	t.printStackTrace();
	if ( ++error_count >= error_max ) {
	System.out.println("Exceeded error count. Exiting.");
	System.exit(1);
	}
	}

	//
	// 1. Instantiate the pinger if possible.
	// 2. Read ducc.proeprties to find the ping interval
	// 3. Start pinging and wriging results to stdout
	//
	// The ServiceManager must start this process as the user. It monitors stdout for success
	// or failute of the ping and reacts accordingly.
	//
	protected void start(String[] args)
	{


	Options options = new Options();
	addOptions(options);

	CommandLineParser parser = new PosixParser();
	CommandLine commandLine = null;
	IServiceStatistics default_statistics = new ServiceStatistics(false, false, "<N/A>");

	try {
	commandLine = parser.parse(options, args);
	} catch (ParseException e) {
	print("Cannot parse command line:", e);
	return;
	}

	String arguments = commandLine.getOptionValue(ServicePing.Arguments.decode());
	String pingClass = commandLine.getOptionValue(ServicePing.Class.decode());
	String endpoint = commandLine.getOptionValue(ServicePing.Endpoint.decode());
	String port = commandLine.getOptionValue(ServicePing.Port.decode());

	Socket sock = null;
	try {
	sock = new Socket("localhost", Integer.parseInt(port));
	} catch (NumberFormatException e2) {
	e2.printStackTrace();
	return;
	} catch (UnknownHostException e2) {
	e2.printStackTrace();
	return;
	} catch (IOException e2) {
	e2.printStackTrace();
	return;
	}
	print ("ServicePingMain listens on port", sock.getLocalPort());
	InputStream sock_in = null;
	OutputStream sock_out = null;
	try {
	sock_in = sock.getInputStream();
	sock_out = sock.getOutputStream();
	} catch (IOException e2) {
	e2.printStackTrace();
	return;
	}

	ObjectOutputStream oos;
	try {
	oos = new ObjectOutputStream(sock_out);
	oos.flush();
	} catch (IOException e1) {
	e1.printStackTrace();
	return;
	}

	AServicePing custom = resolve(pingClass, arguments, endpoint);
	if ( custom == null ) {
	print("bad_pinger:", pingClass, endpoint);
	return;
	}

	while ( true ) {
	if ( debug ) print("ServicePingMeta starts ping.");

	byte[] cmd = new byte[1];
	cmd[0] = 0;
	int eof = 0;
	try {
	eof = sock_in.read(cmd);
	} catch (IOException e) {
	handleError(e);
	}
	if ( debug ) print("Read cmd", new String(cmd), "eof", eof);

	if ( eof == -1 ) {
	print("EOF on input pipe. Exiting");
	custom.stop();
	return;
	}

	try {
	if ( cmd[0] == 'P' ) {
	IServiceStatistics ss = custom.getStatistics();
	if ( ss == null ) {
	ss = default_statistics;
	}
	// print("Is alive: " + ss.isAlive());
	oos.writeObject(ss);
	oos.flush();

	// The ObjectOutputStream will cache instances and if all you do is change a
	// field or two in the object, it won't be detected and the stale object will be
	// sent. So you have to reset() the stream, (or use a new object, or use
	// clone() here also if you want, but this is simplest and safest since we have
	// no control over what the external pinger gives us.
	oos.reset();
	} else if ( cmd[0] == 'Q' ) {
	custom.stop();
	return;
	} else {
	System.err.println("Invalid command recieved: " + Byte.toString(cmd[0]));
	}
	} catch (Throwable e) {
	handleError(e);
	}
	}
	}

	public static void main(String[] args)
	{
	ServicePingMain wrapper = new ServicePingMain();
	wrapper.start(args);
	}

	}