<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 org.seaborne.delta.client;

import static org.seaborne.delta.DeltaConst.VERSION_UNSET ;

import java.util.Objects;
import java.util.concurrent.atomic.AtomicReference ;
import java.util.function.Consumer ;

import org.apache.jena.atlas.lib.Lib ;
import org.apache.jena.atlas.lib.Pair ;
import org.apache.jena.atlas.logging.FmtLog;
import org.apache.jena.atlas.logging.Log;
import org.apache.jena.atlas.web.HttpException;
import org.apache.jena.graph.Node;
import org.apache.jena.query.ReadWrite ;
import org.apache.jena.sparql.core.DatasetGraph;
import org.seaborne.delta.*;
import org.seaborne.delta.link.DeltaLink;
import org.seaborne.delta.link.RegToken;
import org.seaborne.patch.RDFChanges;
import org.seaborne.patch.RDFPatch ;
import org.seaborne.patch.changes.RDFChangesApply ;
import org.seaborne.patch.changes.RDFChangesCollector;
import org.seaborne.patch.system.DatasetGraphChanges;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory ;

/** Provides an interface to a specific dataset over the general {@link DeltaLink} API. 
 * This is the client API, c.f. JDBC connection
 */ 
public class DeltaConnection implements AutoCloseable {
    
    private static Logger LOG = LoggerFactory.getLogger(DeltaConnection.class);//Delta.DELTA_LOG;
    
    // The version of the remote copy.
    private final DeltaLink dLink ;

    // Last seen PatchLogInfo
    private final AtomicReference<PatchLogInfo> remote = new AtomicReference<>(null);
    private final DatasetGraph base;
    private final DatasetGraphChanges managed;
    
    private final RDFChanges target;
    private final String datasourceName ;
    private final Id datasourceId;
    private final DataState state;

    private boolean valid = false;
 
    /** 
     * Connect to an existing {@code DataSource} with the {@link DatasetGraph} as local state.
     * The {@code DatasetGraph} must be in-step with the zone.
     * {@code DeltaConnection} objects are normal obtained via {@link DeltaClient}.
     */
    
    /*package*/ static DeltaConnection connect(DataState dataState, DatasetGraph dsg, DeltaLink dLink) {
        Objects.requireNonNull(dataState, "Null data state");
        Objects.requireNonNull(dataState.getDataSourceId(), "Null data source Id");
        Objects.requireNonNull(dataState.getDatasourceName(), "Null data source Id");
        Objects.requireNonNull(dLink, "Null link");
        DeltaConnection client = new DeltaConnection(dataState, dsg, dLink);
        client.start();
        return client;
    }
    
    private static void ensureRegistered(DeltaLink link, Id clientId) {
        if ( ! link.isRegistered() )
            link.register(clientId);
    }
    
    private DeltaConnection(DataState dataState, DatasetGraph basedsg, DeltaLink link) {
        Objects.requireNonNull(dataState, "DataState");
        Objects.requireNonNull(link, "DeltaLink");
        if ( basedsg instanceof DatasetGraphChanges )
            Log.warn(this.getClass(), "DatasetGraphChanges passed into "+Lib.className(this));
        this.state = dataState;
        this.base = basedsg;
        this.datasourceId = dataState.getDataSourceId();
        this.datasourceName = dataState.getDatasourceName();
        this.dLink = link;
        this.valid = true;
        if ( basedsg != null  ) {
            // Where to put incoming changes. 
            this.target = new RDFChangesApply(basedsg);
            // Where to send outgoing changes.
            RDFChanges monitor = createRDFChanges(datasourceId);
            this.managed = new DatasetGraphChanges(basedsg, monitor, null, syncer());
        } else {
            this.target = null;
            this.managed = null;
        }
    }
    
    // If a read, try but carry on.
    private Consumer<ReadWrite> syncer() {
        return (rw)->{
            try { this.sync() ; }
            catch (RuntimeException ex) {
                if ( rw == ReadWrite.WRITE )
                    throw ex;
            }
        };
    }<commit_msg>Default to sync on W but not R<commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 org.seaborne.delta.client;

import static org.seaborne.delta.DeltaConst.VERSION_UNSET ;

import java.util.Objects;
import java.util.concurrent.atomic.AtomicReference ;
import java.util.function.Consumer ;

import org.apache.jena.atlas.lib.Lib ;
import org.apache.jena.atlas.lib.Pair ;
import org.apache.jena.atlas.logging.FmtLog;
import org.apache.jena.atlas.logging.Log;
import org.apache.jena.atlas.web.HttpException;
import org.apache.jena.graph.Node;
import org.apache.jena.query.ReadWrite ;
import org.apache.jena.sparql.core.DatasetGraph;
import org.seaborne.delta.*;
import org.seaborne.delta.link.DeltaLink;
import org.seaborne.delta.link.RegToken;
import org.seaborne.patch.RDFChanges;
import org.seaborne.patch.RDFPatch ;
import org.seaborne.patch.changes.RDFChangesApply ;
import org.seaborne.patch.changes.RDFChangesCollector;
import org.seaborne.patch.system.DatasetGraphChanges;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory ;

/** Provides an interface to a specific dataset over the general {@link DeltaLink} API. 
 * This is the client API, c.f. JDBC connection
 */ 
public class DeltaConnection implements AutoCloseable {
    
    private static Logger LOG = LoggerFactory.getLogger(DeltaConnection.class);//Delta.DELTA_LOG;
    
    // The version of the remote copy.
    private final DeltaLink dLink ;

    // Last seen PatchLogInfo
    private final AtomicReference<PatchLogInfo> remote = new AtomicReference<>(null);
    private final DatasetGraph base;
    private final DatasetGraphChanges managed;
    
    private final RDFChanges target;
    private final String datasourceName ;
    private final Id datasourceId;
    private final DataState state;

    private boolean valid = false;
 
    /** 
     * Connect to an existing {@code DataSource} with the {@link DatasetGraph} as local state.
     * The {@code DatasetGraph} must be in-step with the zone.
     * {@code DeltaConnection} objects are normal obtained via {@link DeltaClient}.
     */
    
    /*package*/ static DeltaConnection connect(DataState dataState, DatasetGraph dsg, DeltaLink dLink) {
        Objects.requireNonNull(dataState, "Null data state");
        Objects.requireNonNull(dataState.getDataSourceId(), "Null data source Id");
        Objects.requireNonNull(dataState.getDatasourceName(), "Null data source Id");
        Objects.requireNonNull(dLink, "Null link");
        DeltaConnection client = new DeltaConnection(dataState, dsg, dLink);
        client.start();
        return client;
    }
    
    private static void ensureRegistered(DeltaLink link, Id clientId) {
        if ( ! link.isRegistered() )
            link.register(clientId);
    }
    
    private DeltaConnection(DataState dataState, DatasetGraph basedsg, DeltaLink link) {
        Objects.requireNonNull(dataState, "DataState");
        Objects.requireNonNull(link, "DeltaLink");
        if ( basedsg instanceof DatasetGraphChanges )
            Log.warn(this.getClass(), "DatasetGraphChanges passed into "+Lib.className(this));
        this.state = dataState;
        this.base = basedsg;
        this.datasourceId = dataState.getDataSourceId();
        this.datasourceName = dataState.getDatasourceName();
        this.dLink = link;
        this.valid = true;
        if ( basedsg != null  ) {
            // Where to put incoming changes. 
            this.target = new RDFChangesApply(basedsg);
            // Where to send outgoing changes.
            RDFChanges monitor = createRDFChanges(datasourceId);
            this.managed = new DatasetGraphChanges(basedsg, monitor, null, syncer());
        } else {
            this.target = null;
            this.managed = null;
        }
    }
    
    // Sync if write.
    private Consumer<ReadWrite> syncer() {
        return (rw)->{
            switch(rw) {
                case READ: 
                    try { sync(); } catch (Exception ex) {}
                    break;
                case WRITE:
                    this.sync();
                    break;
            }
        };
    }
{'fraction_non_alphanumeric': 0.09027205276174774, 'fraction_numerical': 0.001236603462489695, 'mean_word_length': 2.9839901477832513, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 11, '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': '7252582', 'n_tokens_mistral': 2676, 'n_tokens_neox': 2635, 'n_words': 802}