<commit_before>/* Copyright 2015 <NAME>
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package crud.jdbc;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Objects;
import java.util.concurrent.Callable;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;

import crud.core.DataSource;
import crud.util.SessionWorker;
import rx.Observable;
import rx.Subscriber;


/*package*/ class QueryDataSource implements DataSource<ResultSetRow> {

    private @Nonnull final Connection connection;
    private @Nonnull final StatementTemplate statementTemplate;
    private @Nonnull final SessionWorker worker;

    /**
     * A query to be executed, cached for reuse. Should only be accessed in
     * the {@link #worker}'s thread, so no need to make it volatile or worry
     * about race conditions.
     */
    private @Nullable PreparedStatement statement = null;


    public QueryDataSource(
            @Nonnull final Connection connection,
            @Nonnull final StatementTemplate statementTemplate,
            @Nonnull final SessionWorker worker) {
        this.connection = Objects.requireNonNull(connection);
        this.statementTemplate = Objects.requireNonNull(statementTemplate);
        this.worker = Objects.requireNonNull(worker);
    }

    @Override
    public Observable<ResultSetRow> read() {
        return Observable.create(new Observable.OnSubscribe<ResultSetRow>() {
            @Override
            public void call(final Subscriber<? super ResultSetRow> sub) {
                QueryDataSource.this.worker.submit(new Callable<Void>() {
                    @SuppressWarnings("resource")
                    @Override
                    public Void call() throws Exception {
                        final PreparedStatement raceFreeStmt = getStatement();
                        try (ResultSet results = raceFreeStmt.executeQuery()) {
                            while (results.next()) {
                                final ResultSetRow currentRow = new ResultSetRow(results);
                                sub.onNext(currentRow);
                            }
                            sub.onCompleted();
                        }
                        return null;
                    }
                });
            }
        });
    }
<commit_msg>Fixed bug in QueryDataSource: extra call to onCompleted()<commit_after>/* Copyright 2015 <NAME>
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package crud.jdbc;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Objects;
import java.util.concurrent.Callable;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;

import crud.core.DataSource;
import crud.util.SessionWorker;
import rx.Observable;
import rx.Subscriber;


/*package*/ class QueryDataSource implements DataSource<ResultSetRow> {

    private @Nonnull final Connection connection;
    private @Nonnull final StatementTemplate statementTemplate;
    private @Nonnull final SessionWorker worker;

    /**
     * A query to be executed, cached for reuse. Should only be accessed in
     * the {@link #worker}'s thread, so no need to make it volatile or worry
     * about race conditions.
     */
    private @Nullable PreparedStatement statement = null;


    public QueryDataSource(
            @Nonnull final Connection connection,
            @Nonnull final StatementTemplate statementTemplate,
            @Nonnull final SessionWorker worker) {
        this.connection = Objects.requireNonNull(connection);
        this.statementTemplate = Objects.requireNonNull(statementTemplate);
        this.worker = Objects.requireNonNull(worker);
    }

    @Override
    public Observable<ResultSetRow> read() {
        return Observable.create(new Observable.OnSubscribe<ResultSetRow>() {
            @Override
            public void call(final Subscriber<? super ResultSetRow> sub) {
                QueryDataSource.this.worker.submit(new Callable<Void>() {
                    @SuppressWarnings("resource")
                    @Override
                    public Void call() throws Exception {
                        final PreparedStatement queryStmt = getStatement();
                        try (ResultSet results = queryStmt.executeQuery()) {
                            while (results.next()) {
                                final ResultSetRow currentRow = new ResultSetRow(results);
                                sub.onNext(currentRow);
                            }
                            // Don't call onCompleted(): that's called by submit()
                            //sub.onCompleted();
                        }
                        return null;
                    }
                });
            }
        });
    }

{'fraction_non_alphanumeric': 0.07740408991042758, 'fraction_numerical': 0.002704073009971269, 'mean_word_length': 2.284128745837958, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4323049', 'n_tokens_mistral': 1397, 'n_tokens_neox': 1350, 'n_words': 464}