def each
      # Need to close the connection, teh result_set, AND the result_set.getStatement when
      # we're done.
      connection = open_connection!

      # We're going to need to ask for item/copy info while in the
      # middle of streaming our results. JDBC is happier and more performant
      # if we use a seperate connection for this.
      extra_connection = open_connection! if include_some_holdings?

      # We're going to make our marc records in batches, and only yield
      # them to caller in batches, so we can fetch copy/item info in batches
      # for efficiency.
      batch_size = settings["horizon.batch_size"].to_i
      record_batch = []

      exclude_tags = (settings["horizon.exclude_tags"] || "").split(",")


      rs = self.fetch_result_set!(connection)

      current_bib_id = nil
      record = nil
      record_count = 0

      error_handler = org.marc4j.ErrorHandler.new

      while(rs.next)
        bib_id      = rs.getInt("bib#");

        if bib_id != current_bib_id
          record_count += 1

          if settings["debug_ascii_progress"] &&  (record_count % settings["solrj_writer.batch_size"] == 0)
            $stderr.write ","
          end

          # new record! Put old one on batch queue.
          record_batch << record if record

          # prepare and yield batch?
          if (record_count % batch_size == 0)
            enhance_batch!(extra_connection, record_batch)
            record_batch.each do |r|
              # set current_bib_id for error logging
              current_bib_id = r['001'].value
              yield r
            end
            record_batch.clear
          end

          # And start new record we've encountered.
          error_handler = org.marc4j.ErrorHandler.new
          current_bib_id = bib_id
          record = MARC::Record.new
          record.append MARC::ControlField.new("001", bib_id.to_s)
        end

        tagord      = rs.getInt("tagord");
        tag         = rs.getString("tag")

        # just silently skip it, some weird row in the horizon db, it happens.
        # plus any of our exclude_tags.
        next if tag.nil? || tag == "" || exclude_tags.include?(tag)

        indicators = rs.getString("indicators")

        # a packed byte array could be in various columns, in order of preference...
        # the xref stuff is joined in from the auth table
        # Have to get it as bytes and then convert it to String to avoid JDBC messing
        # up the encoding marc8 grr
        authtext = rs.getBytes("xref_longtext") || rs.getBytes("xref_text")
        text     = rs.getBytes("longtext") || rs.getBytes("text")


        if tag == "000"
          # Horizon puts a \x1E marc field terminator on the end of hte
          # leader in the db too, but that's not really part of it.
          record.leader =  String.from_java_bytes(text).chomp("\x1E")

          fix_leader!(record.leader)
        elsif tag != "001"
          # we add an 001 ourselves with bib id in another part of code.
          field = build_marc_field!(error_handler, tag, indicators, text, authtext)
          record.append field unless field.nil?
        end
      end

      # last one
      record_batch << record if record

      # yield last batch
      enhance_batch!(extra_connection, record_batch)

      record_batch.each do |r|
        yield r
      end
      record_batch.clear

    rescue Exception => e
      logger.fatal "HorizonReader, unexpected exception at bib id:#{current_bib_id}: #{e}"    
      raise e
    ensure
      logger.info("HorizonReader: Closing all JDBC objects...")

      # have to cancel the statement to keep us from waiting on entire
      # result set when exception is raised in the middle of stream.
      statement = rs && rs.getStatement
      if statement
        statement.cancel
        statement.close
      end

      rs.close if rs

      # shouldn't actually need to close the resultset and statement if we cancel, I think.
      connection.close if connection

      extra_connection.close if extra_connection

      logger.info("HorizonReader: Closed JDBC objects")
    end
def each
      # Need to close the connection, teh result_set, AND the result_set.getStatement when
      # we're done.
      connection = open_connection!

      # We're going to need to ask for item/copy info while in the
      # middle of streaming our results. JDBC is happier and more performant
      # if we use a seperate connection for this.
      extra_connection = open_connection! if include_some_holdings?

      # We're going to make our marc records in batches, and only yield
      # them to caller in batches, so we can fetch copy/item info in batches
      # for efficiency.
      batch_size = settings["horizon.batch_size"].to_i
      record_batch = []

      exclude_tags = (settings["horizon.exclude_tags"] || "").split(",")


      rs = self.fetch_result_set!(connection)

      current_bib_id = nil
      record = nil
      record_count = 0

      error_handler = org.marc4j.ErrorHandler.new

      while(rs.next)
        bib_id      = rs.getInt("bib#");

        if bib_id != current_bib_id
          record_count += 1

          if settings["debug_ascii_progress"] &&  (record_count % settings["solrj_writer.batch_size"] == 0)
            $stderr.write ","
          end

          # new record! Put old one on batch queue.
          record_batch << record if record

          # prepare and yield batch?
          if (record_count % batch_size == 0)
            enhance_batch!(extra_connection, record_batch)
            record_batch.each do |r|
              # set current_bib_id for error logging
              current_bib_id = r['001'].value
              yield r
            end
            record_batch.clear
          end

          # And start new record we've encountered.
          error_handler = org.marc4j.ErrorHandler.new
          current_bib_id = bib_id
          record = MARC::Record.new
          record.append MARC::ControlField.new("001", bib_id.to_s)
        end

        tagord      = rs.getInt("tagord");
        tag         = rs.getString("tag")

        # just silently skip it, some weird row in the horizon db, it happens.
        # plus any of our exclude_tags.
        next if tag.nil? || tag == "" || exclude_tags.include?(tag)

        indicators = rs.getString("indicators")

        # a packed byte array could be in various columns, in order of preference...
        # the xref stuff is joined in from the auth table
        # Have to get it as bytes and then convert it to String to avoid JDBC messing
        # up the encoding marc8 grr
        authtext = rs.getBytes("xref_longtext") || rs.getBytes("xref_text")
        text     = rs.getBytes("longtext") || rs.getBytes("text")


        if tag == "000"
          # Horizon puts a \x1E marc field terminator on the end of hte
          # leader in the db too, but that's not really part of it.
          record.leader =  String.from_java_bytes(text).chomp("\x1E")

          fix_leader!(record.leader)
        elsif tag != "001"
          # we add an 001 ourselves with bib id in another part of code.
          field = build_marc_field!(error_handler, tag, indicators, text, authtext)
          record.append field unless field.nil?
        end
      end

      # last one
      record_batch << record if record

      # yield last batch
      enhance_batch!(extra_connection, record_batch)

      record_batch.each do |r|
        yield r
      end
      record_batch.clear

    rescue Exception => e
      logger.fatal "HorizonReader, unexpected exception at bib id:#{current_bib_id}: #{e}"    
      raise e
    ensure
      logger.info("HorizonReader: Closing all JDBC objects...")

      # have to cancel the statement to keep us from waiting on entire
      # result set when exception is raised in the middle of stream.
      statement = rs && rs.getStatement
      if statement
        statement.cancel
        statement.close
      end

      rs.close if rs

      # shouldn't actually need to close the resultset and statement if we cancel, I think.
      connection.close if connection

      extra_connection.close if extra_connection

      logger.info("HorizonReader: Closed JDBC objects")
    end
Read rows from horizon database, assemble them into MARC::Record's, and yield each
 MARC::Record to caller.