InfoSphere Streams, REST API, Java application

Introduction
InfoSphere Streams contains API (application programming interface) which can be used to get access to different data related to Streams instance. The API is based on REST  and is implemented by using HTTP protocol. The format of data returned is JSON. More information is available here (chapter: REST API overview.
The API allows creating applications for monitoring and visualizing data related to the Strems instance and particular job or jobs. The Streams Console (provided with InforSphere Streams) is based  on REST API. More information is available here (chapter: Streams Console )
Authentication
The REST API is supported by SWS (Streams Web Services chapter: Streams Web Services). There is a difference between instance authentication (for instance: to use streamtool) and SWS authentication. There are two methods of SWS authentication: server (default) and client. More information is available here (chapter: Configuring security for the InfoSphere Streams REST API).
Server authentication
It is the default method. Firstly let start Streams Console application from any browser. Streams Console is based on REST API and allows to check if SWS is up and running and authentication is possible.
After starting the instance use the streamtool to get URL for Streams Console.

[streams@oc8442647460 ~]$ streamtool geturl -i test@streams

https://oc8442647460.ibm.com:8443/streams/console/login 

If login to Streams Console is successful we are sure that the access data is valid.
Sample Java code

 static class BusinessIntelligenceX509TrustManager implements
   X509TrustManager {

  public java.security.cert.X509Certificate[] getAcceptedIssuers() {
   return new java.security.cert.X509Certificate[] {};
  }

  public void checkClientTrusted(
    java.security.cert.X509Certificate[] certs, String authType) {
   // no-op
  }

  public void checkServerTrusted(
    java.security.cert.X509Certificate[] certs, String authType) {
   // no-op
  }

 }

 public static void auth1() {
  try {

   TrustManager[] trustAllCerts = new TrustManager[] { new BusinessIntelligenceX509TrustManager() };
   SSLContext sc;

   try {
    sc = SSLContext.getInstance("SSL");
   } catch (NoSuchAlgorithmException noSuchAlgorithmException) {
    return;
   }

   try {
    sc.init(null, trustAllCerts, new java.security.SecureRandom());
   } catch (KeyManagementException keyManagementException) {

    return;
   }

   HostnameVerifier hv = new HostnameVerifier() {
    public boolean verify(String urlHostName, SSLSession session) {
     return true;
    }
   };

   HttpsURLConnection
     .setDefaultSSLSocketFactory(sc.getSocketFactory());
   HttpsURLConnection.setDefaultHostnameVerifier(hv);

   // Retrieve the root resource information for Infosphere Streams
   URL url = new URL("https://st32:8443/streams/rest/resources");
   String userInfo = "streams:secret123";
   String authToken = "Basic "
     + DatatypeConverter.printBase64Binary(userInfo.getBytes());
   HttpsURLConnection conn = (HttpsURLConnection) url.openConnection();
   conn.setRequestProperty("Authorization", authToken);
   conn.setRequestMethod("GET");
   conn.connect();
   System.out.println("Response code: " + conn.getResponseCode());
   System.out.println("Content type: "
     + conn.getHeaderField("Content-Type"));
   String response = new BufferedReader(new InputStreamReader(
     conn.getInputStream())).readLine();
   System.out.println("Response: " + response);
   conn.disconnect();
  } catch (Exception e) {
   e.printStackTrace();
  }
 }


 public static void main(String[] args) {
  auth1();
 }

Client authentication 
If client authentication method is enabled only trusted clients can connect to SWS and REST API. Unfortunately, this method is a little bit more complicated and requires more preparation.
Enable client authentication 
It can be done from Streams Console. But the simplest method is to modify a file .streams/instances/{instanceid}/config/instance.properties and set SWS.enableClientAuthentication property to true.
Create and register client credentials
Create client credentials
Important: if the cn name (first and last name) is the same as instance owner then login name and the and password are not required during authentication.

 keytool -genkey -keyalg RSA -alias streams -storepass secret -validity 360 -keysize 1024 -storetype pkcs12 -keystore streamkey.p12

Add credentials to Web browser
keytool --export -keystore streamkey.p12 -alias streams -storetype pkcs12 -file my.crt
 For instance (Chromium): Properties -> Advanced -> Certificates -> Import certificate
Add certificate to SWS client keystore

streamtool addcertificate --clientid sb -f my.crt -i test@streams

Restart instance

streamtool stopinstance -i test@streams
streamtool startinstance -i test@streams

Streams Console
Open Streams Console again, should start without asking for credentials (if certificate first and last name is the same as instance owner, otherwise login and password is required)
Server certificate 
Next step is to create server trust store at the client site. Unfortunately, there is not visible method for getting server certificate using streamtool. The only solution I found (after Streams Console is successfully opened) is exporting server certificate directly from browser. In case of firefox : Preferences -> Advanced -> List certifacates -> Servers -> IBM branch. Export certificate as X.509 (DERT) and create server truststore.

keytool -import -alias streams -keystore servertrust -file serv.cert

Java sample 
Important: for some reason the client keystore create above does not work here. The only solution I found is to backup client certicate in Web browser (Firefox: Preferences ->Advanced -> List certificates -> Personal -> Backup as PKCS12) and used keystore created that way.
Important: This example assumes that client certificate contains cn (first and last name) which maps to instance owner or valid Streams user, so the authorization is ignored. Otherwise it is necessary to add basic authentication also (like example above).

 static void auth3() {
  try {
   // Identify locations of server truststore and client keystore
   System.setProperty("javax.net.ssl.trustStore",
     "/home/sbartkowski/servertrust");
   System.setProperty("javax.net.ssl.trustStorePassword", "secret");
   System.setProperty("javax.net.ssl.keyStore",
     "/home/sbartkowski/clientkeystore.p12");
   System.setProperty("javax.net.ssl.keyStorePassword", "secret");
   System.setProperty("javax.net.ssl.keyStoreType", "pkcs12");
   System.setProperty("javax.net.debug", "ssl");

   HostnameVerifier hv = new HostnameVerifier() {
    public boolean verify(String urlHostName, SSLSession session) {
     return true;
    }
   };

   HttpsURLConnection.setDefaultHostnameVerifier(hv);

   // Retrieve the root resource information for Infosphere Streams
   URL url = new URL("https://st32:8443/streams/rest/resources");
   HttpsURLConnection conn = (HttpsURLConnection) url.openConnection();
   conn.connect();
   System.out.println("Response code: " + conn.getResponseCode());
   System.out.println("Content type: "
     + conn.getHeaderField("Content-Type"));
   String response = new BufferedReader(new InputStreamReader(
     conn.getInputStream())).readLine();
   System.out.println("Response: " + response);
   conn.disconnect();
  } catch (Exception e) {
   e.printStackTrace();
  }
 }

JavaHotel - reservation panel improved

New version of JavaHotel has been uploaded (U/P user/user). The main difference is improved reservation panel : more compact, better navigation and better visualization of reservation.

The main challenge I fought with was adding colspan to CellTable. After spending a lot of time trying to apply different methods I ended up with copying and pasting DefaultCellTableBuilder and adding necessary changed to the underlying algorithm.

IBM InfoSphere Streams, NetezzaLoad, ODBCAppend and input control operator

Introduction
In the previous post it was demonstrated how to connect to Netezza using connection data in the connections.xml file. But this method in not always convenient. For instance: assume we use different credentials in the test environment and production environment. Having credentials hardcoded in the connections.xml enforces recompiling our application every time it is deployed.
Solution
Happily there is a solution to this problem. It is possible also to send credentials during submit time and even to change them dynamically during a run time. It can be accomplished by applying "control" input stream to ODBCAppend and NetezzaLoad. Description (click on Operator Control Input Port). Unfortunately, there is no sample for this option and it is not obvious at the first glance how to use it.
Sample code
A sample code for connection using control input port is available here.

Remarks

• The producer of connection data is Custom operator submitting credentials to both operators (ODBCAppend and NetezzaLoad). Although the credentials are hardcoded here it is not a problem to get them from somewhere, for instance a configuration file.
• ODBCAppend and NetezzaLoad have a connectionPassword with invalid password hardcoded. It is on purpose to be sure that during a connection a password got from control port is used.
• ODBCAppend and NetezzaLoad have parameter connectionPolicy set to Deferred (default is Immediate). Otherwise both operators would try to initialize connection at the beginning and fail. By means of Deferred value the connection is not started until first tuple arrives.
• It is also necessary to postpone tuples arrival to ODBCAppend and NetezzaLoad until connection credentials are sent. Switch control serves as a gate between flow of tuples and both data operators. Switch operator having parameter start set to true serves as a gate latch. 
• Custom operator sends credentials to ODBCAppend and NetezzaLoad and at the same time sends signal to Switch operator. The gate is broken and data can flow through not interrupted.

How it is visualized in the diagram above

1. Beacon operator starts sending tuples. Tuples are sent (and blocked at the very beginning) to the FlowOfControl (Switch) operator.
2. ControlPort (Custom) operator sends connection data to the ToODBC and NetezzaLoad operator. 
3. After sending the connection data sends a signal to FlowOfControl (Switch) operator to open a gate.
4. At the arrival of first tuple the connection is started using (valid) credentials from control input port (not from connections.xml file or invalid connectionPassword parameter).
5. Then the constant flow of tuples from producer (Beacon) to consumers (ODBAppend and NetezzaLoad) is running. Tuples are passing through FlowOfControl gate without any friction.