<commit_before>/*
 *  The OpenDiamond Platform for Interactive Search
 *  Version 5
 *
 *  Copyright (c) 2009 Carnegie Mellon University
 *  All rights reserved.
 *
 *  This software is distributed under the terms of the Eclipse Public
 *  License, Version 1.0 which can be found in the file named LICENSE.
 *  ANY USE, REPRODUCTION OR DISTRIBUTION OF THIS SOFTWARE CONSTITUTES
 *  RECIPIENT'S ACCEPTANCE OF THIS AGREEMENT
 */

package edu.cmu.cs.diamond.opendiamond;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.Socket;
import java.util.List;
import java.util.Set;

class Connection {

    private static final int DIAMOND_PORT = 5872;

    private static final int NONCE_SIZE = 16;

    final private MiniRPCConnection control;

    final private MiniRPCConnection blast;

    final private String hostname;

    String getHostname() {
        return hostname;
    }

    // all public methods must close() on IOException!

    private static Socket createOneChannel(String address, byte nonce[])
            throws IOException {
        if (nonce.length != NONCE_SIZE) {
            throw new IllegalArgumentException("nonce[] must be NONCE_SIZE ("
                    + NONCE_SIZE + "), actual size " + nonce.length);
        }

        Socket socket = new Socket(address, DIAMOND_PORT);
        // System.out.println(address);
        // System.out.println(socket);
        DataInputStream in = new DataInputStream(socket.getInputStream());
        DataOutputStream out = new DataOutputStream(socket.getOutputStream());

        // write nonce
        // System.out.println("writing " + Arrays.toString(nonce));
        out.write(nonce);

        // read nonce
        in.readFully(nonce);

        // System.out.println("read " + Arrays.toString(nonce));

        return socket;
    }

    Connection(MiniRPCConnection control, MiniRPCConnection blast,
            String hostname) {
        this.control = control;
        this.blast = blast;
        this.hostname = hostname;
    }

    static Connection createConnection(String host, List<Cookie> cookieList,
            Set<String> pushAttributes, XDR_sig_and_data fspec,
            List<Filter> filters) throws ServerException {
        // System.out.println("connecting to " + host);

        byte nonce[] = new byte[NONCE_SIZE];

        MiniRPCConnection control;
        MiniRPCConnection blast;

        try {
            // open control (if exception is thrown here, it's ok)
            control = new MiniRPCConnection(createOneChannel(host, nonce));

            // open data
            try {
                blast = new MiniRPCConnection(createOneChannel(host, nonce));
            } catch (IOException e) {
                try {
                    // close control and propagate
                    control.close();
                } catch (IOException e2) {
                }
                throw e;
            }

            Connection conn = new Connection(control, blast, host);
            conn.sendPreStart(cookieList, pushAttributes, fspec, filters);
            return conn;
        } catch (IOException e) {
            throw new ServerException(host, e);
        }
    }

    // TODO pipeline
    private void sendPreStart(List<Cookie> cookieList,
            Set<String> pushAttributes, XDR_sig_and_data fspec,
            List<Filter> filters) throws IOException {
        try {
            // define scope
            for (Cookie cookie : cookieList) {
                byte[] data = XDREncoders.encodeString(cookie.getCookie());
                new RPC(this, hostname, 24, data).doRPC().checkStatus();
            }

            // set the push attributes
            if (pushAttributes != null) {
                byte[] encodedAttributes = new XDR_attr_name_list(
                        pushAttributes).encode();
                new RPC(this, hostname, 20, encodedAttributes).doRPC()
                        .checkStatus();
            }

            // set the fspec
            // device_set_spec = 6
            new RPC(this, hostname, 6, fspec.encode()).doRPC().checkStatus();

            // set the codes and blobs
            for (Filter f : filters) {
                setCode(f);
                setBlob(f);
            }
        } catch (IOException e) {
            close();
            throw e;
        }
    }

    public void sendStart() throws IOException {
        try {
            // start search
            byte encodedSearchId[] = new byte[4]; // 0

            // device_start_search = 1
            new RPC(this, hostname, 1, encodedSearchId).doRPC().checkStatus();
        } catch (IOException e) {
            close();
            throw e;
        }
    }

    private void setBlob(Filter f) throws IOException {
        final byte encodedBlobSig[] = f.getEncodedBlobSig();
        final byte encodedBlob[] = f.getEncodedBlob();

        // System.out.println("blob sig: " + encodedBlobSig);

        // device_set_blob_by_signature = 22
        MiniRPCReply reply1 = new RPC(this, hostname, 22, encodedBlobSig)
                .doRPC();
        if (reply1.getMessage().getStatus() != RPC.DIAMOND_FCACHEMISS) {
            reply1.checkStatus();
            return;
        }

        // device_set_blob = 11
        new RPC(this, hostname, 11, encodedBlob).doRPC().checkStatus();
    }

    private void setCode(Filter f) throws IOException {
        FilterCode code = f.getFilterCode();
        XDR_sig_val sig = new XDR_sig_val(code.getSignature());
        XDR_sig_and_data sigAndData = new XDR_sig_and_data(sig,
                code.getBytes());

        final byte[] encodedSig = sig.encode();
        final byte[] encodedSigAndData = sigAndData.encode();

        // device_set_obj = 16
        MiniRPCReply reply1 = new RPC(this, hostname, 16, encodedSig).doRPC();
        if (reply1.getMessage().getStatus() != RPC.DIAMOND_FCACHEMISS) {
            reply1.checkStatus();
            return;
        }

        // device_send_obj = 17
        new RPC(this, hostname, 17, encodedSigAndData).doRPC().checkStatus();
    }

    void close() {
        // System.out.println("closing " + toString());

        try {
            control.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        try {
            blast.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public MiniRPCMessage receiveBlast() throws ServerException {
        try {
            return blast.receive();
        } catch (IOException e) {
            close();
            throw new ServerException(hostname, e);
        }
    }

    public void sendBlastRequest(int cmd, byte data[]) throws ServerException {
        try {
            blast.sendRequest(cmd, data);
        } catch (IOException e) {
            close();
            throw new ServerException(hostname, e);
        }
    }

    public void sendControlRequest(int cmd, byte[] data) throws ServerException {
        try {
            control.sendRequest(cmd, data);
        } catch (IOException e) {
            close();
            throw new ServerException(hostname, e);
        }
    }

    public MiniRPCMessage receiveControl() throws ServerException {
        try {
            return control.receive();
        } catch (IOException e) {
            close();
            throw new ServerException(hostname, e);
        }
    }
}
<commit_msg>Send non-zero search_id with device_start_search<commit_after>/*
 *  The OpenDiamond Platform for Interactive Search
 *  Version 5
 *
 *  Copyright (c) 2009 Carnegie Mellon University
 *  All rights reserved.
 *
 *  This software is distributed under the terms of the Eclipse Public
 *  License, Version 1.0 which can be found in the file named LICENSE.
 *  ANY USE, REPRODUCTION OR DISTRIBUTION OF THIS SOFTWARE CONSTITUTES
 *  RECIPIENT'S ACCEPTANCE OF THIS AGREEMENT
 */

package edu.cmu.cs.diamond.opendiamond;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.security.SecureRandom;
import java.util.List;
import java.util.Set;

class Connection {

    private static final int DIAMOND_PORT = 5872;

    private static final int NONCE_SIZE = 16;

    final private MiniRPCConnection control;

    final private MiniRPCConnection blast;

    final private String hostname;

    String getHostname() {
        return hostname;
    }

    // all public methods must close() on IOException!

    private static Socket createOneChannel(String address, byte nonce[])
            throws IOException {
        if (nonce.length != NONCE_SIZE) {
            throw new IllegalArgumentException("nonce[] must be NONCE_SIZE ("
                    + NONCE_SIZE + "), actual size " + nonce.length);
        }

        Socket socket = new Socket(address, DIAMOND_PORT);
        // System.out.println(address);
        // System.out.println(socket);
        DataInputStream in = new DataInputStream(socket.getInputStream());
        DataOutputStream out = new DataOutputStream(socket.getOutputStream());

        // write nonce
        // System.out.println("writing " + Arrays.toString(nonce));
        out.write(nonce);

        // read nonce
        in.readFully(nonce);

        // System.out.println("read " + Arrays.toString(nonce));

        return socket;
    }

    Connection(MiniRPCConnection control, MiniRPCConnection blast,
            String hostname) {
        this.control = control;
        this.blast = blast;
        this.hostname = hostname;
    }

    static Connection createConnection(String host, List<Cookie> cookieList,
            Set<String> pushAttributes, XDR_sig_and_data fspec,
            List<Filter> filters) throws ServerException {
        // System.out.println("connecting to " + host);

        byte nonce[] = new byte[NONCE_SIZE];

        MiniRPCConnection control;
        MiniRPCConnection blast;

        try {
            // open control (if exception is thrown here, it's ok)
            control = new MiniRPCConnection(createOneChannel(host, nonce));

            // open data
            try {
                blast = new MiniRPCConnection(createOneChannel(host, nonce));
            } catch (IOException e) {
                try {
                    // close control and propagate
                    control.close();
                } catch (IOException e2) {
                }
                throw e;
            }

            Connection conn = new Connection(control, blast, host);
            conn.sendPreStart(cookieList, pushAttributes, fspec, filters);
            return conn;
        } catch (IOException e) {
            throw new ServerException(host, e);
        }
    }

    // TODO pipeline
    private void sendPreStart(List<Cookie> cookieList,
            Set<String> pushAttributes, XDR_sig_and_data fspec,
            List<Filter> filters) throws IOException {
        try {
            // define scope
            for (Cookie cookie : cookieList) {
                byte[] data = XDREncoders.encodeString(cookie.getCookie());
                new RPC(this, hostname, 24, data).doRPC().checkStatus();
            }

            // set the push attributes
            if (pushAttributes != null) {
                byte[] encodedAttributes = new XDR_attr_name_list(
                        pushAttributes).encode();
                new RPC(this, hostname, 20, encodedAttributes).doRPC()
                        .checkStatus();
            }

            // set the fspec
            // device_set_spec = 6
            new RPC(this, hostname, 6, fspec.encode()).doRPC().checkStatus();

            // set the codes and blobs
            for (Filter f : filters) {
                setCode(f);
                setBlob(f);
            }
        } catch (IOException e) {
            close();
            throw e;
        }
    }

    public void sendStart() throws IOException {
        try {
            // Generate a search ID that should be unique over the lifetime
            // of this scope cookie.  OpenDiamond-Java doesn't use this for
            // anything, but the servers may choose to use it (in concert
            // with the scope cookie) to correlate a particular search across
            // multiple servers.  The protocol only gives us 32 bits, so
            // divide them as follows:
            // - 24 bits: millisecond-granularity timestamp right-shifted
            //   by 11 bits.  This gives us a 397-day rollover and a 2-second
            //   resolution.  Most scope cookies expire much sooner than
            //   397 days.
            // - 8 bits: random uniquifier.  This can't be a serial number
            //   because there may be multiple uncoordinated clients.
            //   This allows on average 16 new searches per 2-second tick
            //   before we hit a collision due to the birthday paradox.
            int searchId = (int) (((System.currentTimeMillis() >>> 11)
                    & 0xffffff) << 8);
            byte rand[] = new byte[1];
            new SecureRandom().nextBytes(rand);
            searchId |= ((int) rand[0]) & 0xff;
            byte encodedSearchId[] = new byte[4];
            ByteBuffer bb = ByteBuffer.allocate(4);
            bb.putInt(searchId).flip();
            bb.get(encodedSearchId);

            // device_start_search = 1
            new RPC(this, hostname, 1, encodedSearchId).doRPC().checkStatus();
        } catch (IOException e) {
            close();
            throw e;
        }
    }

    private void setBlob(Filter f) throws IOException {
        final byte encodedBlobSig[] = f.getEncodedBlobSig();
        final byte encodedBlob[] = f.getEncodedBlob();

        // System.out.println("blob sig: " + encodedBlobSig);

        // device_set_blob_by_signature = 22
        MiniRPCReply reply1 = new RPC(this, hostname, 22, encodedBlobSig)
                .doRPC();
        if (reply1.getMessage().getStatus() != RPC.DIAMOND_FCACHEMISS) {
            reply1.checkStatus();
            return;
        }

        // device_set_blob = 11
        new RPC(this, hostname, 11, encodedBlob).doRPC().checkStatus();
    }

    private void setCode(Filter f) throws IOException {
        FilterCode code = f.getFilterCode();
        XDR_sig_val sig = new XDR_sig_val(code.getSignature());
        XDR_sig_and_data sigAndData = new XDR_sig_and_data(sig,
                code.getBytes());

        final byte[] encodedSig = sig.encode();
        final byte[] encodedSigAndData = sigAndData.encode();

        // device_set_obj = 16
        MiniRPCReply reply1 = new RPC(this, hostname, 16, encodedSig).doRPC();
        if (reply1.getMessage().getStatus() != RPC.DIAMOND_FCACHEMISS) {
            reply1.checkStatus();
            return;
        }

        // device_send_obj = 17
        new RPC(this, hostname, 17, encodedSigAndData).doRPC().checkStatus();
    }

    void close() {
        // System.out.println("closing " + toString());

        try {
            control.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        try {
            blast.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public MiniRPCMessage receiveBlast() throws ServerException {
        try {
            return blast.receive();
        } catch (IOException e) {
            close();
            throw new ServerException(hostname, e);
        }
    }

    public void sendBlastRequest(int cmd, byte data[]) throws ServerException {
        try {
            blast.sendRequest(cmd, data);
        } catch (IOException e) {
            close();
            throw new ServerException(hostname, e);
        }
    }

    public void sendControlRequest(int cmd, byte[] data) throws ServerException {
        try {
            control.sendRequest(cmd, data);
        } catch (IOException e) {
            close();
            throw new ServerException(hostname, e);
        }
    }

    public MiniRPCMessage receiveControl() throws ServerException {
        try {
            return control.receive();
        } catch (IOException e) {
            close();
            throw new ServerException(hostname, e);
        }
    }
}

{'fraction_non_alphanumeric': 0.09664948453608248, 'fraction_numerical': 0.007118311242022582, 'mean_word_length': 1.8747574528135473, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 18, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '5520760', 'n_tokens_mistral': 4459, 'n_tokens_neox': 4216, 'n_words': 1182}