Scalikejdbc result set итератор

Question

Я использую scalikejdbc для доступа к огромной таблице. Мое понимание - он извлекает все строки в память, прежде чем я смогу отобразить или перебрать их.

В настоящее время у меня есть реализация с использованием rxscala Observable, что очень просто. Но приемник медленнее, чем чтение sql, а затем я получаю OutOfMemory из-за буферизации. Вот мой нынешний продюсер, как мог Observable:

def fetchProductsAsObservable(
    sql: SQL[Nothing,NoExtractor], 
    extractor: (WrappedResultSet) => ProductItem) 
) = 
    Observable[ProductItem](o => 
     try { 
      sql.foreach(row => o.onNext(extractor(row))) 
      o.onCompleted() 
     } catch { 
     case e: Throwable => o.onError(e) 
     } 
    ) 

Я знаю метод SQL.foreach, но он получает метод обратного вызова и возвращает блок. Мой фон в .NET. Не могу понять, как правильно реализовать простой Iterator в scala с помощью scalikejdbc, который я могу дать для параллельной обработки?

Answer 1

Ваш вопрос не связан с поведением ScalikeJDBC, а с RxJava. Поскольку ваш Observable дополнен горячим источником, вам нужно использовать стратегию противодавления.