(Map<String, Object> topoConf, TopologyContext context,
SpoutOutputCollector collector)
| 90 | } |
| 91 | |
| 92 | @Override |
| 93 | public void open(Map<String, Object> topoConf, TopologyContext context, |
| 94 | SpoutOutputCollector collector) { |
| 95 | this.collector = collector; |
| 96 | this.context = context; |
| 97 | |
| 98 | if (topoConf.containsKey(Config.TOPOLOGY_SUBPROCESS_TIMEOUT_SECS)) { |
| 99 | workerTimeoutMills = 1000 * ObjectReader.getInt(topoConf.get(Config.TOPOLOGY_SUBPROCESS_TIMEOUT_SECS)); |
| 100 | } else { |
| 101 | workerTimeoutMills = 1000 * ObjectReader.getInt(topoConf.get(Config.SUPERVISOR_WORKER_TIMEOUT_SECS)); |
| 102 | } |
| 103 | |
| 104 | process = new ShellProcess(command); |
| 105 | if (!env.isEmpty()) { |
| 106 | process.setEnv(env); |
| 107 | } |
| 108 | |
| 109 | Number subpid = process.launch(topoConf, context, changeDirectory); |
| 110 | LOG.info("Launched subprocess with pid " + subpid); |
| 111 | |
| 112 | logHandler = ShellUtils.getLogHandler(topoConf); |
| 113 | logHandler.setUpContext(ShellSpout.class, process, this.context); |
| 114 | |
| 115 | heartBeatExecutorService = MoreExecutors.getExitingScheduledExecutorService(new ScheduledThreadPoolExecutor(1)); |
| 116 | } |
| 117 | |
| 118 | @Override |
| 119 | public void close() { |
nothing calls this directly
no test coverage detected