<commit_before>//----------------------------------------------------------------------
//This code is developed as part of the Java CoG Kit project
//The terms of the license can be found at http://www.cogkit.org/license
//This message may not be removed or altered.
//----------------------------------------------------------------------

/*
 * Created on Jan 1, 2010
 */
package org.globus.cog.abstraction.impl.file.coaster.handlers.providers;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

import org.apache.log4j.Logger;
import org.globus.cog.abstraction.impl.file.coaster.buffers.Buffers;
import org.globus.cog.abstraction.impl.file.coaster.buffers.Buffers.Direction;
import org.globus.cog.abstraction.impl.file.coaster.buffers.ReadBuffer;
import org.globus.cog.abstraction.impl.file.coaster.buffers.ReadBufferCallback;
import org.globus.cog.abstraction.impl.file.coaster.buffers.ThrottleManager;
import org.globus.cog.abstraction.impl.file.coaster.buffers.WriteBuffer;
import org.globus.cog.abstraction.impl.file.coaster.buffers.WriteBufferCallback;
import org.globus.cog.abstraction.impl.file.coaster.handlers.CoasterFileRequestHandler;

public class LocalIOProvider implements IOProvider {
    public static final Logger logger = Logger.getLogger(LocalIOProvider.class); 

    public void abort(IOHandle handle) throws IOException {
        ((Abortable) handle).abort();
    }

    public IOReader pull(String src, String dest, ReadIOCallback cb) throws IOException {
        return new Reader(src, cb);
    }

    public IOWriter push(String src, String dest, WriteIOCallback cb) throws IOException {
        return new Writer(dest, cb);
    }

    public boolean isDirect() {
        return true;
    }
    
    private static URI newURI(String src) throws IOException {
        try {
            return new URI(src);
        }
        catch (URISyntaxException e) {
            throw new IOException("Malformed URI: " + e.getMessage());
        }
    }

    private static class Writer implements IOWriter, WriteBufferCallback, Abortable {
        private static Direction BUFDIR = Direction.OUT;
        
        private File f;
        private long len, crt;
        private WriteIOCallback cb;
        private WriteBuffer buf;

        public Writer(String dest, WriteIOCallback cb) throws IOException {
            this.cb = cb;
            URI destURI = newURI(dest);
            f = CoasterFileRequestHandler.normalize(destURI.getPath().substring(1));
        }
        
        public String toString() {
            return "LW " + f;
        }

        public void close() throws IOException {
            buf.close();
        }

        public void setLength(long len) throws IOException {
            this.len = len;
            if (len == 0) {
                f.createNewFile();
                cb.done(this);
            }
            else {
                File p = f.getParentFile();
                if (!p.exists()) {
                    if (!p.mkdirs()) {
                        throw new IOException("Failed to create directory " + p.getAbsolutePath());
                    }
                }
                buf = Buffers.newWriteBuffer(Buffers.getBuffers(Direction.OUT), new FileOutputStream(f).getChannel(), this);
            }
        }

        public void write(boolean last, byte[] data) throws IOException {
            crt += data.length;
            if (last && crt != len) {
                throw new IOException(" File size mismatch. Expected " + len + " bytes, got "
                        + crt);
            }
            try {
                buf.write(last, data);
            }
            catch (InterruptedException e) {
                throw new IOException("Interrupted!");
            }
        }

        public void done(boolean last) {
            if (last) {
                cb.done(this);
            }
        }

        public void error(boolean last, Exception e) {
            cb.error(this, e);
        }

        public void abort() throws IOException {
            buf.close();
            f.delete();
        }

        public void suspend() {
            // not used
        }

        public void resume() {
            // not used
        }
        <commit_msg>avoid NPE if an error happens before the transfer starts<commit_after>//----------------------------------------------------------------------
//This code is developed as part of the Java CoG Kit project
//The terms of the license can be found at http://www.cogkit.org/license
//This message may not be removed or altered.
//----------------------------------------------------------------------

/*
 * Created on Jan 1, 2010
 */
package org.globus.cog.abstraction.impl.file.coaster.handlers.providers;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

import org.apache.log4j.Logger;
import org.globus.cog.abstraction.impl.file.coaster.buffers.Buffers;
import org.globus.cog.abstraction.impl.file.coaster.buffers.Buffers.Direction;
import org.globus.cog.abstraction.impl.file.coaster.buffers.ReadBuffer;
import org.globus.cog.abstraction.impl.file.coaster.buffers.ReadBufferCallback;
import org.globus.cog.abstraction.impl.file.coaster.buffers.ThrottleManager;
import org.globus.cog.abstraction.impl.file.coaster.buffers.WriteBuffer;
import org.globus.cog.abstraction.impl.file.coaster.buffers.WriteBufferCallback;
import org.globus.cog.abstraction.impl.file.coaster.handlers.CoasterFileRequestHandler;

public class LocalIOProvider implements IOProvider {
    public static final Logger logger = Logger.getLogger(LocalIOProvider.class); 

    public void abort(IOHandle handle) throws IOException {
        ((Abortable) handle).abort();
    }

    public IOReader pull(String src, String dest, ReadIOCallback cb) throws IOException {
        return new Reader(src, cb);
    }

    public IOWriter push(String src, String dest, WriteIOCallback cb) throws IOException {
        return new Writer(dest, cb);
    }

    public boolean isDirect() {
        return true;
    }
    
    private static URI newURI(String src) throws IOException {
        try {
            return new URI(src);
        }
        catch (URISyntaxException e) {
            throw new IOException("Malformed URI: " + e.getMessage());
        }
    }

    private static class Writer implements IOWriter, WriteBufferCallback, Abortable {
        private static Direction BUFDIR = Direction.OUT;
        
        private File f;
        private long len, crt;
        private WriteIOCallback cb;
        private WriteBuffer buf;

        public Writer(String dest, WriteIOCallback cb) throws IOException {
            this.cb = cb;
            URI destURI = newURI(dest);
            f = CoasterFileRequestHandler.normalize(destURI.getPath().substring(1));
        }
        
        public String toString() {
            return "LW " + f;
        }

        public void close() throws IOException {
            buf.close();
        }

        public void setLength(long len) throws IOException {
            this.len = len;
            if (len == 0) {
                f.createNewFile();
                cb.done(this);
            }
            else {
                File p = f.getParentFile();
                if (!p.exists()) {
                    if (!p.mkdirs()) {
                        throw new IOException("Failed to create directory " + p.getAbsolutePath());
                    }
                }
                buf = Buffers.newWriteBuffer(Buffers.getBuffers(Direction.OUT), new FileOutputStream(f).getChannel(), this);
            }
        }

        public void write(boolean last, byte[] data) throws IOException {
            crt += data.length;
            if (last && crt != len) {
                throw new IOException(" File size mismatch. Expected " + len + " bytes, got "
                        + crt);
            }
            try {
                buf.write(last, data);
            }
            catch (InterruptedException e) {
                throw new IOException("Interrupted!");
            }
        }

        public void done(boolean last) {
            if (last) {
                cb.done(this);
            }
        }

        public void error(boolean last, Exception e) {
            cb.error(this, e);
        }

        public void abort() throws IOException {
            if (buf != null) {
                buf.close();
            }
            f.delete();
        }

        public void suspend() {
            // not used
        }

        public void resume() {
            // not used
        }
        
{'fraction_non_alphanumeric': 0.1252236135957066, 'fraction_numerical': 0.0017889087656529517, 'mean_word_length': 1.7866043613707165, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '426009', 'n_tokens_mistral': 2413, 'n_tokens_neox': 2364, 'n_words': 600}