@Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        // Save parameters from storm
        this.conf = conf;
        this.context = context;
        this.collector = collector;
        // get cursor, where we left of, for this executor
        // get this task id
        //this.query=getOpsLogQuery(context.getThisComponentId());
        // Set up an executor
        // if clustered sharded setup, use multiple threads
        // connect to mongoS first and get sharded configuration information
        boolean sharded=isSharded();
        MongoURI uri = new MongoURI(this.url);
        Mongo mongo = null;
        ArrayList<String>  hosts=new ArrayList<String>();
        try {
            mongo = new Mongo(uri);
        } catch (UnknownHostException e) {
            e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
        }





        if (sharded) { // get each server and port info from shards collection
            DB db = mongo.getDB("config");
            DBCursor cursor=db.getCollection("shards").find();
            while( cursor.hasNext() )  {
                DBObject obj = cursor.next();
                String host=obj.get("host").toString();
                String replSet=host.split("/")[0];
                hosts.add(host.split("/")[1]);
                List addrs = new ArrayList();
                for (String replicaSetHosts: hosts) {

                    String[] individualHost= replicaSetHosts.split(",");
                    for (int i = 0; i < individualHost.length ; i++) {
                        String[] HostPort = individualHost[i].split(":");

                        try {
                            addrs.add(new ServerAddress( HostPort[0],Integer.parseInt(HostPort[1])) );
                        } catch (UnknownHostException e) {
                            e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
                        }


                    }
                    // create a thread for each ReplicaSet in the shard
                    this.spoutTask = new MongoSpoutTask(this.queue, addrs, this.dbName, this.collectionNames, this.query);
                    // Start thread
                    Thread thread = new Thread(this.spoutTask);
                    thread.start();

                    addrs.clear();
                }
            }
        }else {
            // there is just simple replica set configuration
            // replica set information stored in
            List addrs = new ArrayList();
            DB db = mongo.getDB("admin");
            DBObject cmd = new BasicDBObject();
            cmd.put("replSetGetStatus", 1);
            CommandResult result = db.command(cmd);
            ArrayList<BasicDBObject> members= (ArrayList<BasicDBObject>) result.get("members");
            for (int i = 0; i < members.size(); i++) {
                BasicDBObject o =  members.get(i);
                String[] hostPort=o.getString("name").split(":");
                try {
                    addrs.add( new ServerAddress( hostPort[0] , Integer.parseInt(hostPort[1]) ) );
                } catch (UnknownHostException e) {
                    e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
                }
            }
            // create a thread with set of addresses
            // create a thread for each ReplicaSet in the shard
            this.spoutTask = new MongoSpoutTask(this.queue, addrs, this.dbName, this.collectionNames, this.query);
            // Start thread
            Thread thread = new Thread(this.spoutTask);
            thread.start();


        }






    }