Hazelcast Jet ReadJdbcP Sınıfı - JDBC Source ResultSet Üzerinde Yürür ve Emit Eder

Giriş

Şu satırı dahil ederiz

import com.hazelcast.jet.impl.connector.ReadJdbcP;

Processor açısından bakarsak en önemli metodlar şöyle

1. init()

Yeni bir JDBC Connection nesnesi alır

2. complete()

SQL cümlesini çalıştıran ve bir ResultSet elde eden yer burası.  Elde edilen ResultSet üzerinde yürünür ve her bir satır için mapOutputFn çağrılır ve döndürülen nesnesiyi emit eder. Yani Outbox'a yazar

3. close()

JDBC ResultSet, JDBC Statement , JDBC Connection nesnelerini kapatır

constructor

Kod şöyle. Bu sınıf SelectProcessorSupplier, SourceProcessors ve SecuredFunctions tarafından yaratılır. 

İkinci parametre resultSetFn olarak isimlendiriliyor. PreparedStatement ile SQL cümlesini çalıştıran kod

Üçüncü parametre mapOutputFn olarak isimlendiriliyor ancak aslında ismi rowMapper olmalıydı. complete() metodunda kullanılır

public ReadJdbcP(
  SupplierEx<? extends Connection> newConnectionFn,
  ToResultSetFunction resultSetFn,
  FunctionEx<? super ResultSet, ? extends T> mapOutputFn
) {
  ...
}

close metodu

JDBC ResultSet, JDBC Statement , JDBC Connection gibi kaynakları kapatır

complete metodu

resultSetFn.createResultSet() çağrısı ile SQL cümlesini çalıştıran ve bir ResultSet elde eden yer burası.  Elde edilen ResultSet üzerinde yürünür ve her bir satır için mapOutputFn çağrılır. Kod şöyle.

@Override
public boolean complete() {
  if (traverser == null) {
    resultSet = uncheckCall(() -> resultSetFn
      .createResultSet(connection, parallelism, index));
    traverser = ((Traverser<ResultSet>) () -> uncheckCall(() -> resultSet.next() ?
                                                                resultSet : null))
      .map(mapOutputFn);
  
  return emitFromTraverser(traverser);
}

init metodu

Yeni bir JDBC Connection nesnesi alır

supplier metodu

İmzaları şöyle

public static <T> ProcessorMetaSupplier supplier(
  SupplierEx<? extends DataSource> newDataSourceFn,
  ToResultSetFunction resultSetFn,
  FunctionEx<? super ResultSet, ? extends T> mapOutputFn
)

public static <T> ProcessorMetaSupplier supplier(
  FunctionEx<ProcessorSupplier.Context, ? extends Connection> newConnectionFn,
  ToResultSetFunction resultSetFn,
  FunctionEx<? super ResultSet, ? extends T> mapOutputFn
) 

public static <T> ProcessorMetaSupplier supplier(
  String connectionURL,
  String query,
  FunctionEx<? super ResultSet, ? extends T> mapOutputFn
)

public static <T> ProcessorMetaSupplier supplier(
  DataConnectionRef dataConnectionRef,
  ToResultSetFunction resultSetFn,
  FunctionEx<? super ResultSet, ? extends T> mapOutputFn)

ReadJdbcP için static factory metodlarıdır. Sources.jdbc(..) çağrıları için processor üretilen yer burasıdır.