@Override
  public void prepare(final Map conf, final TopologyContext context, final OutputCollector collector) {
    _ruby_bolt = initialize_ruby_bolt();
    IRubyObject ruby_conf = JavaUtil.convertJavaToRuby(__ruby__, conf);
    IRubyObject ruby_context = JavaUtil.convertJavaToRuby(__ruby__, context);