package io.reactiverse.pgclient.impl;
import io.reactiverse.pgclient.PgException;
import io.reactiverse.pgclient.Row;
import io.reactiverse.pgclient.impl.codec.decoder.ErrorResponse;
import io.reactiverse.pgclient.impl.codec.decoder.RowDescription;
import java.util.stream.Collector;
public abstract class QueryCommandBase<T> extends CommandBase<Boolean> {
public RowResultDecoder<?, T> decoder;
final QueryResultHandler<T> resultHandler;
final Collector<Row, ?, T> collector;
QueryCommandBase(Collector<Row, ?, T> collector, QueryResultHandler<T> resultHandler) {
this.resultHandler = resultHandler;
this.collector = collector;
}
abstract String sql();
@Override
public void handleCommandComplete(int updated) {
this.result = false;
T result;
int size;
RowDescription desc;
if (decoder != null) {
result = decoder.complete();
desc = decoder.description();
size = decoder.size();
decoder.reset();
} else {
result = emptyResult(collector);
size = 0;
desc = null;
}
resultHandler.handleResult(updated, size, desc, result);
}
@Override
public void handleErrorResponse(ErrorResponse errorResponse) {
failure = new PgException(errorResponse);
}
private static <A, T> T emptyResult(Collector<Row, A, T> collector) {
return collector.finisher().apply(collector.supplier().get());
}
}