protected void doPrepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector
      collector, Collection<Event<Tuple>> queue) {
    Objects.requireNonNull(topoConf);
    Objects.requireNonNull(context);
    Objects.requireNonNull(collector);
    Objects.requireNonNull(queue);
    this.windowedOutputCollector = new WindowedOutputCollector(collector);
    bolt.prepare(topoConf, context, windowedOutputCollector);
    this.listener = newWindowLifecycleListener();
    this.windowManager = initWindowManager(listener, topoConf, context, queue);
    start();
    LOG.info(String.format("Initialized window manager %s", windowManager));
  }
protected void doPrepare ( Map < String , Object > topoConf , TopologyContext context , OutputCollector collector , Collection < Event < Tuple > > queue ) { Objects . requireNonNull ( topoConf ) ; Objects . requireNonNull ( context ) ; Objects . requireNonNull ( collector ) ; Objects . requireNonNull ( queue ) ; this . windowedOutputCollector = new WindowedOutputCollector ( collector ) ; bolt . prepare ( topoConf , context , windowedOutputCollector ) ; this . listener = newWindowLifecycleListener ( ) ; this . windowManager = initWindowManager ( listener , topoConf , context , queue ) ; start ( ) ; LOG . info ( String . format ( "Initialized window manager %s" , windowManager ) ) ; }
NOTE: the queue has to be thread safe.