/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package sample.tswebclient.cassandra;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.Deletion;
import org.apache.cassandra.thrift.Mutation;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.commons.pool.impl.GenericObjectPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * <p>Cassandra EJB Manager. This is a example of the Terminal Services class but
 * transformed in a new EJB 3.1 Singleton Enterprise Java Bean. The EJB uses
 * thrift and apache commons-pool to communicate with Cassandra in a pooled way.
 * </p>
 *
 * <p>This example EJB is used to store in cassandra an application structure
 * very similar than a common File System. An application can be a folder
 * or a common application. A folder can have more applications (or folder)
 * below. Each application has a name, icon, description, rdp startup file and
 * the type.</p>
 *
 * In order to store this information in cassandra two columns were defined:<br/>
 *
 * <p>
 * <b>ColumnFamily: Applications.</b><br/>
 * It stores the appliaction data for start terminal server applications.<br/>
 * <ul>
 *   <li>Key => complete path uppercase (Ex: "/APPLICATIONS/INTERNET/SKYPE"</li>
 *   <li>Values => The following data:</li>
 *   <ul>
 *     <li>name: application name to show ("Skype").</li>
 *     <li>description: applications description.</li>
 *     <li>icon: 16x16 PNG icon.</li>
 *     <li>type: "app" or "folder".</li>
 *     <li>path: real path in the windows TS ?? (only for apps). NOT USED.</li>
 *     <li>rdp: rdp file to startup this application (only for apps).</li>
 *   </ul>
 * </ul>
 *
 * <pre>
 * Applications: { // CF
 *   /INTERNET: {
 *     name: Internet
 *     description: Internet applications
 *     icon: ...
 *     type: folder
 *   },
 *   /INTERNET/SKYPE: { // key
 *     name: Skype
 *     description: VoIP client application (www.skype.com)
 *     icon: ...
 *     type: app
 *     rdp: ...
 *   }
 * }
 * </pre>
 * </p>
 *
 * <p><b>ColumnFamily: ApplicationStructure</b><br/>
 * It stores the folder structure directories.<br/>
 * <ul>
 *   <li>Key => complete path uppercase (Ex: "/APPLICATIONS").</li>
 *   <li>Values => The list of children applications for this one.</li>
 *   <ul>
 *     <li>name: Children key, "/APPLICATIONS/INTERNET" or "/APPLICATIONS/PAINT".</li>
 *     <li>value: Children application name.</li>
 *   </ul>
 * </ul>
 *
 * <pre>
 * ApplicationStructure: { // CF
 *   /APPLICATIONS: {
 *     /APPLICATIONS/INTERNET: Internet
 *     /APPLICATIONS/PAINT: Paint
 *   },
 *   /APPLICATIONS/INTERNET: { // key
 *     /APPLICATIONS/INTERNET/PIDGIN: Pidgin
 *     /APPLICATIONS/INTERNET/SKYPE: Skype
 *   }
 * }
 * </pre>
 * </p>
 *
 * <p>The class is a simple CRUD manager for applications, with the methods:
 * <ul>
 *   <li>readApplication: the method reads a application passing the key (path)
 *       and returns a complete application. It can be used to read only the
 *       application, the application with its children or recursively read
 *       all applications behind it.</li>
 *   <li>createApplication: Method to create a new application (folder or app),
 *       it gives some errors if the parent does not exist, if a folder
 *       is created with children,...</li>
 *   <li>deleteApplication: Method that deletes a empty folder or application
 *       with the path or key passed as argument. Same as create it throws a 
 *       exception if some conditions (path does not exists, folder with 
 *       children,...). </li>
 * </ul></p>
 *
 * <p>I have also used CDI (Context Dependency Injection) in order to expose the
 * EJB to JSF beans. So use Inject annotation to use this EJB.</p>
 *
 * <p>All the properties that are used to configure the EJB are treated as
 * Resources. This way can be annotated or defined in an ejb-jar.xml file. In
 * this example the host resource is initialized with "cassandra.com" but then
 * redefined in the xml file to "localhost". See the WEB-INF/ejb-jar.xml file
 * to see how it was done.</p>
 *
 * @see Application
 * @author ricky
 */

@javax.inject.Named
@javax.enterprise.context.ApplicationScoped
@javax.ejb.Singleton
@javax.ejb.LocalBean
public class CassandraManagerEJB {
    
    //
    // PROPERTIES & RESOURCES
    //

    private static final Logger logger = LoggerFactory.getLogger(CassandraManagerEJB.class);
    
    @Resource
    private String keyspace = "Keyspace1";

    @Resource
    private String applicationColumnFamily = "Applications";

    @Resource
    private String structureColumnFamily = "ApplicationStructure";

    @Resource
    private String host = "cassandra.com";

    @Resource
    int port = 9160;
    
    private GenericObjectPool pool = null;

    //
    // CONSTRUCTORS
    //

    /**
     * Empty constructor that be used by EJB container.
     */
    public CassandraManagerEJB() {
        // empty for EJB
        logger.debug("Empty Constructor...");
    }

    /**
     * Manual constructor to be used in tests.
     *
     * @param keyspace Cassandra keyspace to use.
     * @param applicationColumnFamily Column for applications.
     * @param structureColumnFamily Column for structure.
     * @param factory The pool factory to use.
     */
    protected CassandraManagerEJB(String keyspace, String applicationColumnFamily,
            String structureColumnFamily, CassandraClientObjectFactory factory) {
        this.keyspace = keyspace;
        this.applicationColumnFamily = applicationColumnFamily;
        this.structureColumnFamily = structureColumnFamily;
        this.host = factory.getHost();
        this.port = factory.getPort();
        this.pool = new GenericObjectPool(factory);
    }

    /**
     * Another constructor for tests.
     *
     * @param keyspace Cassandra keyspace to use.
     * @param applicationColumnFamily Column for applications.
     * @param structureColumnFamily Column for structure.
     * @param factory The pool factory to use.
     * @param config Configuration for the pool if non default values are required.
     */
    protected CassandraManagerEJB(String keyspace, String applicationColumnFamily,
            String structureColumnFamily,
            CassandraClientObjectFactory factory, GenericObjectPool.Config config) {
        this.keyspace = keyspace;
        this.applicationColumnFamily = applicationColumnFamily;
        this.structureColumnFamily = structureColumnFamily;
        this.host = factory.getHost();
        this.port = factory.getPort();
        this.pool = new GenericObjectPool(factory, config);
    }

    //
    // INIT
    //

    /**
     * The post construct method that initializes the Cassandra connection pool.
     */
    @PostConstruct
    public void postConstruct() {
        logger.debug("initPool... " + host + ":" + port);
        this.pool = new GenericObjectPool(new CassandraClientObjectFactory(host, port));
    }

    //
    // GETTERS & SETTERS
    //

    /**
     * Return cassandra application column family used
     * @return Cassandra application column family.
     */
    public String getApplicationColumnFamily() {
        return applicationColumnFamily;
    }

    /**
     * Set a new cassandra application column family
     * @param applicationColumnFamily The new Cassandra application column family
     */
    public void setApplicationColumnFamily(String applicationColumnFamily) {
        this.applicationColumnFamily = applicationColumnFamily;
    }

    /**
     * Return the Cassandra host used
     * @return Cassandra host used
     */
    public String getHost() {
        return host;
    }

    /**
     * Set a new cassandra hostname
     * @param host The new cassandra hostname
     */
    public void setHost(String host) {
        this.host = host;
    }

    /**
     * Return the cassandra keyspace used
     * @return The cassandra keyspace used
     */
    public String getKeyspace() {
        return keyspace;
    }

    /**
     * Set a new cassandra keyspace
     * @param keyspace The new cassandra keyspace
     */
    public void setKeyspace(String keyspace) {
        this.keyspace = keyspace;
    }

    /**
     * Return the cassandra port used
     * @return The cassandra port used
     */
    public int getPort() {
        return port;
    }

    /**
     * Set the new cassandra port
     * @param port The new cassandra port
     */
    public void setPort(int port) {
        this.port = port;
    }

    /**
     * return the cassandra column for structure
     * @return The cassandra column used for structure
     */
    public String getStructureColumnFamily() {
        return structureColumnFamily;
    }

    /**
     * Set a new cassandra column for structure
     * @param structureColumnFamily The new cassandra column for structure info.
     */
    public void setStructureColumnFamily(String structureColumnFamily) {
        this.structureColumnFamily = structureColumnFamily;
    }

    //
    // PRIVATE CASSANDRA METHODS
    //

    /**
     * This method transforms a list of cassandra columns into a list of
     * applications. This method is used when the children of a folder is read
     * in the structure column. As structure column only has key and name the
     * children applications only has the key and name filled.
     *
     * @param cols The columns read in cassandra from structure column for a specified
     *             application folder.
     * @return The list of children application (only key and name properties filled).
     * @throws UnsupportedEncodingException Error in character set with cassandra.
     */
    private List<Application> constructChildren(List<ColumnOrSuperColumn> cols) throws UnsupportedEncodingException {
        List<Application> result = new ArrayList<Application>();
        for (ColumnOrSuperColumn cosc : cols) {
            Column column = cosc.column;
            String childKey = new String(column.name, "UTF8");
            String childName = new String(column.value, "UTF8");
            result.add(new Application(childKey, childName, null));
        }
        return result;
    }

    /**
     * Private method that construct a mutation for for a given name and
     * value and the current time as timestamp.
     *
     * @param name The name for the mutation.
     * @param value The value for the mutation.
     * @return The new mutation constructed.
     */
    private Mutation constructMutation(byte[] name, byte[] value) {
        Column col = new Column();
        col.setName(name);
        col.setValue(value);
        col.setTimestamp(System.currentTimeMillis());
        ColumnOrSuperColumn cosc = new ColumnOrSuperColumn();
        cosc.setColumn(col);
        Mutation mut = new Mutation();
        mut.setColumn_or_supercolumn(cosc);
        return mut;
    }

    /**
     * This private method construct the list of mutations for an application
     * data update or creation. The list of mutations are intended to be applied
     * to the Application cassandra column (not structure).
     *
     * @param app The new application to create mutations.
     * @return The list of mutations to apply.
     * @throws UnsupportedEncodingException Some error in character set.
     */
    private List<Mutation> constructMutationForData(Application app) throws UnsupportedEncodingException {
        List<Mutation> appMut = new ArrayList<Mutation>(5);
        if (app.getName() != null) {
            appMut.add(constructMutation(Attributes.name.getBytes("UTF8"), app.getName().getBytes("UTF8")));
        } else {
            appMut.add(constructMutation(Attributes.name.getBytes("UTF8"), new byte[0]));
        }
        if (app.getType() != null) {
            appMut.add(constructMutation(Attributes.type.getBytes("UTF8"), app.getType().toString().getBytes("UTF8")));
        } else {
            appMut.add(constructMutation(Attributes.type.getBytes("UTF8"), new byte[0]));
        }
        if (app.getIcon() != null) {
            appMut.add(constructMutation(Attributes.icon.getBytes("UTF8"), app.getIcon()));
        } else {
            appMut.add(constructMutation(Attributes.icon.getBytes("UTF8"), new byte[0]));
        }
        if (app.getDescription() != null) {
            appMut.add(constructMutation(Attributes.description.getBytes("UTF8"), app.getDescription().getBytes("UTF8")));
        } else {
            appMut.add(constructMutation(Attributes.description.getBytes("UTF8"), new byte[0]));
        }
        if (app.getPath() != null) {
            appMut.add(constructMutation(Attributes.path.getBytes("UTF8"), app.getPath().getBytes("UTF8")));
        } else {
            appMut.add(constructMutation(Attributes.path.getBytes("UTF8"), new byte[0]));
        }
        if (app.getRdp() != null) {
            appMut.add(constructMutation(Attributes.rdp.getBytes("UTF8"), app.getRdp()));
        } else {
            appMut.add(constructMutation(Attributes.rdp.getBytes("UTF8"), new byte[0]));
        }
        return appMut;
    }

    /**
     * Same method than previous but for structure column. So this method generates
     * all the mutations for a folder creation. This is not used cos createApplication
     * is not allowed with children.
     *
     * @param app The new application folder to create structure mutations.
     * @return The List of mutations to apply.
     * @throws UnsupportedEncodingException Some error with character set.
     */
    private List<Mutation> constructMutationStructure(Application app) throws UnsupportedEncodingException {
        List<Mutation> chMut = new ArrayList<Mutation>(5);
        if (Application.Type.FOLDER.equals(app.getType()) && app.getChildren().size() > 0) {
            // has children
            for (Application child : app.getChildren()) {
                if (child.getKey() != null && child.getType() != null) {
                    chMut.add(constructMutation(child.getKey().getBytes("UTF8"), child.getName().toString().getBytes("UTF8")));
                }
            }
        }
        return chMut;
    }

    /**
     * Private method that constructs the mutations for structure column in a
     * creation. The list of mutations is added to passed muts map. Of course
     * the mutation consists in adding the new application in the specified
     * parent on structure column.
     *
     * @param muts The current mutations map.
     * @param app The new application to be added.
     * @return The mutations map with the addition.
     * @throws UnsupportedEncodingException Some error in encoding.
     */
    private Map<String, Map<String, List<Mutation>>> constructAddChildMutation(
            Map<String, Map<String, List<Mutation>>> muts,
            Application app) throws UnsupportedEncodingException {
        List<Mutation> addChild = new ArrayList<Mutation>(1);
        addChild.add(constructMutation(app.getKey().getBytes("UTF8"), app.getName().getBytes("UTF8")));
        Map<String, List<Mutation>> map = new HashMap<String, List<Mutation>>(1);
        map.put(structureColumnFamily, addChild);
        muts.put(app.getParentKey(), map);
        return muts;
    }

    /**
     * Private method that is used to construct the mutations map to delete
     * an application from its parent when the first one is removed. Of course
     * the mutation is placed in structure column.
     *
     * @param muts The current mutations map.
     * @param app The application to be deleted.
     * @return The mutations map with the deletion added.
     * @throws UnsupportedEncodingException
     */
    private Map<String, Map<String, List<Mutation>>> constructDeleteChildMutation(
            Map<String, Map<String, List<Mutation>>> muts,
            Application app) throws UnsupportedEncodingException {
        SlicePredicate predicate = new SlicePredicate();
        predicate.addToColumn_names(app.getKey().getBytes("UTF8"));
        Deletion deletion = new Deletion();
        deletion.setPredicate(predicate);
        deletion.setTimestamp(System.currentTimeMillis());
        Mutation mutation = new Mutation();
        mutation.setDeletion(deletion);
        List<Mutation> delChild = new ArrayList<Mutation>(1);
        delChild.add(mutation);
        Map<String, List<Mutation>> map = new HashMap<String, List<Mutation>>(1);
        map.put(structureColumnFamily, delChild);
        muts.put(app.getParentKey(), map);
        return muts;
    }

    /**
     * Method that uses the two previous methods to construct mutations for
     * both columns applications and structure.
     *
     * @param muts The current mutations map.
     * @param app The application to create.
     * @return The same mutations argument with mutations for creation added.
     * @throws UnsupportedEncodingException Some error in character set.
     */
    private Map<String, Map<String, List<Mutation>>> constructAddMutation(
            Map<String, Map<String, List<Mutation>>> muts, Application app) throws UnsupportedEncodingException {
        List<Mutation> data = constructMutationForData(app);
        List<Mutation> structure = constructMutationStructure(app);
        Map<String, List<Mutation>> map = new HashMap<String, List<Mutation>>(1);
        map.put(applicationColumnFamily, data);
        if (structure != null && structure.size() > 0) {
            map.put(structureColumnFamily, structure);
        }
        muts.put(app.getKey(), map);
        return muts;
    }

    /**
     * Method that constructs the mutations for a data deletion. So the mutations
     * for delete all application column slices are added. Delete mutation
     * uses column names cos slices are now not supported in thrift.
     *
     * @param muts The actual mutations map.
     * @param app The application to delete data from applications column.
     * @return The mutations map with delete ones added.
     * @throws UnsupportedEncodingException Some error in character set.
     */
    private Map<String, Map<String, List<Mutation>>> constructDeleteMutation(
            Map<String, Map<String, List<Mutation>>> muts, Application app) throws UnsupportedEncodingException {
        SlicePredicate predicate = new SlicePredicate();
        predicate.addToColumn_names(Attributes.name.getBytes("UTF8"));
        predicate.addToColumn_names(Attributes.type.getBytes("UTF8"));
        predicate.addToColumn_names(Attributes.icon.getBytes("UTF8"));
        predicate.addToColumn_names(Attributes.description.getBytes("UTF8"));
        predicate.addToColumn_names(Attributes.path.getBytes("UTF8"));
        predicate.addToColumn_names(Attributes.rdp.getBytes("UTF8"));
        Deletion deletion = new Deletion();
        deletion.setPredicate(predicate);
        deletion.setTimestamp(System.currentTimeMillis());
        Mutation mutation = new Mutation();
        mutation.setDeletion(deletion);
        List<Mutation> delData = new ArrayList<Mutation>(1);
        delData.add(mutation);
        Map<String, List<Mutation>> map = new HashMap<String, List<Mutation>>(1);
        map.put(applicationColumnFamily, delData);
        muts.put(app.getKey(), map);
        return muts;
    }

    /**
     * Private method used to transform a list of columns read from application
     * cassandra column in an Application object. Each column is read and added
     * to the application property.
     *
     * @param key The key of the application read.
     * @param cols The cassandra columns read for this key in applications column.
     * @return The Application object.
     * @throws UnsupportedEncodingException Some error in character set.
     */
    private Application createApplication(String key, List<ColumnOrSuperColumn> cols) throws UnsupportedEncodingException {
        Application app = new Application(key);
        for (ColumnOrSuperColumn cosc : cols) {
            Column column = cosc.column;
            String colName = new String(column.name, "UTF8");
            if (Attributes.name.equals(Attributes.valueOf(colName)) && column.value.length > 0) {
                app.setName(new String(column.value, "UTF8"));
            } else if (Attributes.type.equals(Attributes.valueOf(colName)) && column.value.length > 0) {
                app.setType(Application.Type.valueOf(new String(column.value, "UTF8")));
            } else if (Attributes.description.equals(Attributes.valueOf(colName)) && column.value.length > 0) {
                app.setDescription(new String(column.value, "UTF8"));
            } else if (Attributes.icon.equals(Attributes.valueOf(colName)) && column.value.length > 0) {
                app.setIcon(column.value);
            } else if (Attributes.path.equals(Attributes.valueOf(colName)) && column.value.length > 0) {
                app.setPath(new String(column.value, "UTF8"));
            } else if (Attributes.rdp.equals(Attributes.valueOf(colName)) && column.value.length > 0) {
                app.setRdp(column.value);
            }
        }
        return app;
    }

    //
    // PUBLIC METHODS
    //

    /**
     * The readApplication is the public method to read an application or folder
     * for a specified key. This method first read the application column for
     * the specified key (name, type, description, icon,...) and then the
     * structure column is read (if needed). Depending on readOptions this
     * method can recursively read another application or not.
     *
     * @param key The application key to read ("/APPLICATIONS/SKYPE" for example).
     * @param readOptions Read options to apply.
     * @param attrs The attributes to read (caller can specify the slices which
     *              are read, if null passed all are read).
     * @return The application read from cassandra.
     * @throws Exception Some error.
     */
    public Application readApplication(String key, ChildrenReadOption readOptions,
            Attributes... attrs) throws Exception {
        Cassandra.Client client = (Cassandra.Client) pool.borrowObject();
        try {
            SlicePredicate predicate = new SlicePredicate();
            if (attrs == null || attrs.length == 0) {
                SliceRange sliceRange = new SliceRange();
                sliceRange.setStart(new byte[0]);
                sliceRange.setFinish(new byte[0]);
                predicate.setSlice_range(sliceRange);
            } else {
                // always read name and type
                predicate.addToColumn_names(Attributes.name.getBytes("UTF8"));
                predicate.addToColumn_names(Attributes.type.getBytes("UTF8"));
                for (int i = 0; i < attrs.length; i++) {
                    if (!attrs[i].equals(Attributes.name) && !attrs[i].equals(Attributes.type)) {
                        predicate.addToColumn_names(attrs[i].getBytes("UTF8"));
                    }
                }
            }
            ColumnParent parent = new ColumnParent(applicationColumnFamily);
            List<ColumnOrSuperColumn> cols = client.get_slice(keyspace,
                    key, parent, predicate, ConsistencyLevel.ONE);
            Application app = null;
            if (cols.size() > 0) {
                app = createApplication(key, cols);
                if (readOptions.compareTo(ChildrenReadOption.NONE) > 0
                        && Application.Type.FOLDER.equals(app.getType())) {
                    // read children
                    parent = new ColumnParent(structureColumnFamily);
                    SlicePredicate predicateChildren = new SlicePredicate();
                    SliceRange sliceRange = new SliceRange();
                    sliceRange.setStart(new byte[0]);
                    sliceRange.setFinish(new byte[0]);
                    predicateChildren.setSlice_range(sliceRange);
                    cols = client.get_slice(keyspace,
                            key, parent, predicateChildren, ConsistencyLevel.ONE);
                    List<Application> children = constructChildren(cols);
                    if (readOptions.compareTo(ChildrenReadOption.ONLY_NAMES) > 0) {
                        for (Application child : children) {
                            app.getChildren().add(
                                    readApplication(child.getKey(),
                                    readOptions.equals(ChildrenReadOption.RECURSIVE)
                                    ? ChildrenReadOption.RECURSIVE : ChildrenReadOption.NONE,
                                    attrs));
                        }
                    } else {
                        app.getChildren().addAll(children);
                    }
                }
            }
            return app;
        } finally {
            pool.returnObject(client);
        }
    }

    /**
     * Method to create or update an application or folder. Application name
     * cannot contain "/" character (used in keys).
     *
     * @param app The new or modified application.
     * @throws Exception The application passed has children, the parent specified
     *                   does not exist or is null or some error in cassandra.
     */
    public void createApplication(Application app) throws Exception {
        Cassandra.Client client = (Cassandra.Client) pool.borrowObject();
        try {
            if (app.getName().indexOf("/") != -1) {
                throw new Exception("Application name cannot contains character '/'.");
            }
            if (app.getChildren().size() > 0) {
                throw new Exception("Cannot create a new folder with children... Create one by one.");
            }
            // check parent
            Application parent = readApplication(app.getParentKey(),
                    ChildrenReadOption.NONE,
                    Attributes.name, Attributes.type);
            if (parent == null && !app.equals(Application.root())) {
                throw new Exception("Application folder not found: " + parent.getKey() + ".");
            }
            // real creation
            Map<String, Map<String, List<Mutation>>> muts = new HashMap<String, Map<String, List<Mutation>>>();
            if (!app.equals(Application.root())) {
                constructAddChildMutation(muts, app);
            }
            constructAddMutation(muts, app);
            client.batch_mutate(keyspace, muts, ConsistencyLevel.ANY);
        } finally {
            pool.returnObject(client);
        }
    }

    /**
     * Public method to delete an application or folder. Only the key is
     * used. If the specified key is a folder with children the deletion is not
     * permitted.
     *
     * @param key The application key to delete.
     * @throws Exception Some error.
     */
    public void deleteApplication(String key) throws Exception {
        Cassandra.Client client = (Cassandra.Client) pool.borrowObject();
        try {
            Application app = readApplication(key, ChildrenReadOption.ONLY_NAMES,
                    Attributes.name, Attributes.type);
            if (app == null) {
                throw new Exception("Application " + key + " does not exists.");
            } else if (Application.Type.FOLDER.equals(app.getType())
                    && app.getChildren().size() > 0) {
                throw new Exception("Folder " + key + " is not empty.");
            }
            Map<String, Map<String, List<Mutation>>> muts = new HashMap<String, Map<String, List<Mutation>>>();
            constructDeleteChildMutation(muts, app);
            constructDeleteMutation(muts, app);
            client.batch_mutate(keyspace, muts, ConsistencyLevel.ANY);
        } finally {
            pool.returnObject(client);
        }
    }

    /**
     * Private helper method to read the icon png file.
     *
     * @param name The png file name.
     * @return The bytes to insert in cassandra.
     * @throws FileNotFoundException The icon file does not exist.
     * @throws IOException IO error reading the file.
     */
    static private byte[] readFile(String name) throws FileNotFoundException, IOException {
        byte[] file = new byte[4096];
        InputStream input = new FileInputStream("/home/ricky/NetBeansProjects/tswebclient/extras/" + name);
        int length = input.read(file);
        if (length > 0 && length < file.length) {
            file = Arrays.copyOfRange(file, 0, length);
        } else {
            throw new IOException("File too big!!!");
        }
        return file;
    }

    //
    // TEST MAIN
    //

    /**
     * Test main.
     * 
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        CassandraManagerEJB manager = new CassandraManagerEJB("Keyspace1", "Applications",
                "ApplicationStructure",
                new CassandraClientObjectFactory("localhost", 9160));
        // initialization
        Application root = Application.root();
        // internet directory
        Application internet = new Application(root, "Internet", Application.Type.FOLDER,
                null, "Example application folder", null, null);
        Application skype = new Application(internet, "Skype", Application.Type.APPLICATION,
                readFile("skype.png"), "Skype application.", null, null);
        Application pidgin = new Application(internet, "Pidgin", Application.Type.APPLICATION,
                readFile("pidgin.png"), "Pidgin application.", null, null);
        // util directory
        Application util = new Application(root, "Util", Application.Type.FOLDER,
                null, "Util folder", null, null);
        Application paint = new Application(util, "Paint", Application.Type.APPLICATION,
                readFile("paint.png"),
                "Creates and edits drawings, and displays and edits scanned photos.", null,
                readFile("paint2.rdp"));
        Application notepad = new Application(util, "Notepad", Application.Type.APPLICATION,
                readFile("notepad.png"),
                "Creates and edits text files using basic text formatting.", null,
                readFile("notepad2.rdp"));
        Application wordpad = new Application(util, "WordPad", Application.Type.APPLICATION,
                readFile("wordpad.png"),
                "Creates and edits text documents with complex formatting.", null,
                readFile("wordpad2.rdp"));
        // Office directory
        Application office = new Application(root, "Office", Application.Type.FOLDER,
                null, "Folder for office applications.", null, null);
        Application acroread = new Application(office, "Acrobat Reader", Application.Type.APPLICATION,
                readFile("acroread.png"),
                "Easily view, print, and collaborate on PDF files with free Adobe Reader 9 software.", null,
                null);
        Application openoofice = new Application(office, "OpenOffice", Application.Type.FOLDER,
                null, "OpenOffice software suite", null, null);
        Application oowriter = new Application(openoofice, "Writer", Application.Type.APPLICATION,
                readFile("openofficeorg3-writer.png"),
                "Create and edit text and graphics in letters, reports, documents and Web pages by using Writer.", null,
                null);
        Application oocalc = new Application(openoofice, "Calc", Application.Type.APPLICATION,
                readFile("openofficeorg3-calc.png"),
                "Perform calculation, analyze information and manage lists in spreadsheets by using Calc.", null,
                null);
        Application microsoftoffice = new Application(office, "Microsoft Office", Application.Type.FOLDER,
                null, "Microsoft Office software suite", null, null);
        // Data Bases
        Application ddbb = new Application(root, "DDBB", Application.Type.FOLDER,
                null, "Custom DataBase programs.", null, null);


        //manager.createApplication(root);
        //manager.createApplication(internet);
        //manager.createApplication(skype);
        //manager.createApplication(pidgin);
        //manager.createApplication(util);
        //manager.createApplication(paint);
        //manager.createApplication(notepad);
        //manager.createApplication(wordpad);
        //manager.createApplication(office);
        //manager.createApplication(acroread);
        //manager.createApplication(openoofice);
        //manager.createApplication(oowriter);
        //manager.createApplication(oocalc);
        //manager.createApplication(microsoftoffice);
        //manager.createApplication(ddbb);

        Application test = new Application(Application.root(), "Applications", Application.Type.FOLDER,
                null, "Descripción de applications", null, null);
        //manager.createApplication(test);
        manager.deleteApplication(test.getKey());

        root = manager.readApplication(root.getKey(), ChildrenReadOption.RECURSIVE);
        internet = manager.readApplication(internet.getKey(), ChildrenReadOption.ONLY_NAMES);
        skype = manager.readApplication(pidgin.getKey(), ChildrenReadOption.NONE);

        System.err.print(root.printPath());
        System.err.println(manager.readApplication(test.getKey(), ChildrenReadOption.NONE));
    }
}