Method prepare

storm-client/src/jvm/org/apache/storm/task/ShellBolt.java:128–174

(Map<String, Object> topoConf, TopologyContext context,
                        final OutputCollector collector)

    }

    @Override
    public void prepare(Map<String, Object> topoConf, TopologyContext context,
                        final OutputCollector collector) {
        if (ConfigUtils.isLocalMode(topoConf)) {
            isLocalMode = true;
        }
        Object maxPending = topoConf.get(Config.TOPOLOGY_SHELLBOLT_MAX_PENDING);
        if (maxPending != null) {
            this.pendingWrites = new ShellBoltMessageQueue(((Number) maxPending).intValue());
        }

        rand = new Random();
        this.collector = collector;

        this.context = context;

        if (topoConf.containsKey(Config.TOPOLOGY_SUBPROCESS_TIMEOUT_SECS)) {
            workerTimeoutMills = 1000 * ObjectReader.getInt(topoConf.get(Config.TOPOLOGY_SUBPROCESS_TIMEOUT_SECS));
        } else {
            workerTimeoutMills = 1000 * ObjectReader.getInt(topoConf.get(Config.SUPERVISOR_WORKER_TIMEOUT_SECS));
        }

        process = new ShellProcess(command);
        if (!env.isEmpty()) {
            process.setEnv(env);
        }

        //subprocesses must send their pid first thing
        Number subpid = process.launch(topoConf, context, changeDirectory);
        LOG.info("Launched subprocess with pid " + subpid);

        logHandler = ShellUtils.getLogHandler(topoConf);
        logHandler.setUpContext(ShellBolt.class, process, this.context);

        // reader
        readerThread = new Thread(new BoltReaderRunnable());
        readerThread.start();

        writerThread = new Thread(new BoltWriterRunnable());
        writerThread.start();

        LOG.info("Start checking heartbeat...");
        setHeartbeat();

        heartBeatExecutorService = MoreExecutors.getExitingScheduledExecutorService(new ScheduledThreadPoolExecutor(1));
        heartBeatExecutorService.scheduleAtFixedRate(new BoltHeartbeatTimerTask(this), 1, 1, TimeUnit.SECONDS);
    }

    @Override
    public void execute(Tuple input) {