/**
 * Tell the Swift server to expect a multi-part upload by submitting
 * a 0-byte file with the X-Object-Manifest header
 *
 * @param path path of final final
 * @throws IOException
 */
public void createManifestForPartUpload(Path path) throws IOException {
  String pathString = toObjectPath(path).toString();
  if (!pathString.endsWith("/")) {
    pathString = pathString.concat("/");
  }
  if (pathString.startsWith("/")) {
    pathString = pathString.substring(1);
  }

  swiftRestClient.upload(toObjectPath(path),
          new ByteArrayInputStream(new byte[0]),
          0,
          new Header(SwiftProtocolConstants.X_OBJECT_MANIFEST, pathString));
}
 

/**
 * Get a configuration which a small blocksize reported to callers
 * @return a configuration for this test
 */
@Override
public Configuration getConf() {
  Configuration conf = super.getConf();
  /*
   * set to 4KB
   */
  conf.setInt(SwiftProtocolConstants.SWIFT_BLOCKSIZE, SWIFT_READ_BLOCKSIZE);
  return conf;
}
 

/**
 * Create a config with a 1KB request size
 * @return a config
 */
@Override
protected Configuration createConfiguration() {
  Configuration conf = super.createConfiguration();
  conf.set(SwiftProtocolConstants.SWIFT_REQUEST_SIZE, "1");
  return conf;
}
 

@Override
protected Configuration createConfiguration() {
  Configuration conf = super.createConfiguration();
  //set the partition size to 1 KB
  conf.setInt(SwiftProtocolConstants.SWIFT_PARTITION_SIZE, PART_SIZE);
  return conf;
}
 

@Override
@Before
public void setUp() throws Exception {
  Configuration conf = new Configuration();
  //small blocksize for faster remote tests
  conf.setInt(SwiftProtocolConstants.SWIFT_BLOCKSIZE, 2);
  URI serviceURI = SwiftTestUtils.getServiceURI(conf);
  fSys = FileSystem.get(serviceURI, conf);
  super.setUp();
}
 

private void assertLocationValid(BlockLocation location) throws
                                                         IOException {
  LOG.info(location);
  String[] hosts = location.getHosts();
  String[] names = location.getNames();
  assertNotEqual("No hosts supplied for " + location, 0, hosts.length);
  //for every host, there's a name.
  assertEquals("Unequal names and hosts in " + location,
               hosts.length, names.length);
  assertEquals(SwiftProtocolConstants.BLOCK_LOCATION,
               location.getNames()[0]);
  assertEquals(SwiftProtocolConstants.TOPOLOGY_PATH,
               location.getTopologyPaths()[0]);
}
 

/**
 * Tell the Swift server to expect a multi-part upload by submitting
 * a 0-byte file with the X-Object-Manifest header
 *
 * @param path path of final final
 * @throws IOException
 */
public void createManifestForPartUpload(Path path) throws IOException {
  String pathString = toObjectPath(path).toString();
  if (!pathString.endsWith("/")) {
    pathString = pathString.concat("/");
  }
  if (pathString.startsWith("/")) {
    pathString = pathString.substring(1);
  }

  swiftRestClient.upload(toObjectPath(path),
          new ByteArrayInputStream(new byte[0]),
          0,
          new Header(SwiftProtocolConstants.X_OBJECT_MANIFEST, pathString));
}
 

/**
 * Get a configuration which a small blocksize reported to callers
 * @return a configuration for this test
 */
@Override
public Configuration getConf() {
  Configuration conf = super.getConf();
  /*
   * set to 4KB
   */
  conf.setInt(SwiftProtocolConstants.SWIFT_BLOCKSIZE, SWIFT_READ_BLOCKSIZE);
  return conf;
}
 

/**
 * Create a config with a 1KB request size
 * @return a config
 */
@Override
protected Configuration createConfiguration() {
  Configuration conf = super.createConfiguration();
  conf.set(SwiftProtocolConstants.SWIFT_REQUEST_SIZE, "1");
  return conf;
}
 

@Override
protected Configuration createConfiguration() {
  Configuration conf = super.createConfiguration();
  //set the partition size to 1 KB
  conf.setInt(SwiftProtocolConstants.SWIFT_PARTITION_SIZE, PART_SIZE);
  return conf;
}
 

@Override
@Before
public void setUp() throws Exception {
  Configuration conf = new Configuration();
  //small blocksize for faster remote tests
  conf.setInt(SwiftProtocolConstants.SWIFT_BLOCKSIZE, 2);
  URI serviceURI = SwiftTestUtils.getServiceURI(conf);
  fSys = FileSystem.get(serviceURI, conf);
  super.setUp();
}
 

private void assertLocationValid(BlockLocation location) throws
                                                         IOException {
  LOG.info(location);
  String[] hosts = location.getHosts();
  String[] names = location.getNames();
  assertNotEqual("No hosts supplied for " + location, 0, hosts.length);
  //for every host, there's a name.
  assertEquals("Unequal names and hosts in " + location,
               hosts.length, names.length);
  assertEquals(SwiftProtocolConstants.BLOCK_LOCATION,
               location.getNames()[0]);
  assertEquals(SwiftProtocolConstants.TOPOLOGY_PATH,
               location.getTopologyPaths()[0]);
}
 

/**
 * Tell the Swift server to expect a multi-part upload by submitting
 * a 0-byte file with the X-Object-Manifest header
 *
 * @param path path of final final
 * @throws IOException
 */
public void createManifestForPartUpload(Path path) throws IOException {
  String pathString = toObjectPath(path).toString();
  if (!pathString.endsWith("/")) {
    pathString = pathString.concat("/");
  }
  if (pathString.startsWith("/")) {
    pathString = pathString.substring(1);
  }

  swiftRestClient.upload(toObjectPath(path),
          new ByteArrayInputStream(new byte[0]),
          0,
          new Header(SwiftProtocolConstants.X_OBJECT_MANIFEST, pathString));
}
 

/**
 * Get a configuration which a small blocksize reported to callers
 * @return a configuration for this test
 */
@Override
public Configuration getConf() {
  Configuration conf = super.getConf();
  /*
   * set to 4KB
   */
  conf.setInt(SwiftProtocolConstants.SWIFT_BLOCKSIZE, SWIFT_READ_BLOCKSIZE);
  return conf;
}
 

/**
 * Create a config with a 1KB request size
 * @return a config
 */
@Override
protected Configuration createConfiguration() {
  Configuration conf = super.createConfiguration();
  conf.set(SwiftProtocolConstants.SWIFT_REQUEST_SIZE, "1");
  return conf;
}
 

@Override
protected Configuration createConfiguration() {
  Configuration conf = super.createConfiguration();
  //set the partition size to 1 KB
  conf.setInt(SwiftProtocolConstants.SWIFT_PARTITION_SIZE, PART_SIZE);
  return conf;
}
 

private void assertLocationValid(BlockLocation location) throws
                                                         IOException {
  LOG.info(location);
  String[] hosts = location.getHosts();
  String[] names = location.getNames();
  assertNotEqual("No hosts supplied for " + location, 0, hosts.length);
  //for every host, there's a name.
  assertEquals("Unequal names and hosts in " + location,
               hosts.length, names.length);
  assertEquals(SwiftProtocolConstants.BLOCK_LOCATION,
               location.getNames()[0]);
  assertEquals(SwiftProtocolConstants.TOPOLOGY_PATH,
               location.getTopologyPaths()[0]);
}
 

/**
 * Return an array containing hostnames, offset and size of
 * portions of the given file.  For a nonexistent
 * file or regions, null will be returned.
 * <p>
 * This call is most helpful with DFS, where it returns
 * hostnames of machines that contain the given file.
 * <p>
 * The FileSystem will simply return an elt containing 'localhost'.
 */
@Override
public BlockLocation[] getFileBlockLocations(FileStatus file,
                                             long start,
                                             long len) throws IOException {
  //argument checks
  if (file == null) {
    return null;
  }

  if (start < 0 || len < 0) {
    throw new IllegalArgumentException("Negative start or len parameter" +
                                       " to getFileBlockLocations");
  }
  if (file.getLen() <= start) {
    return new BlockLocation[0];
  }

  // Check if requested file in Swift is more than 5Gb. In this case
  // each block has its own location -which may be determinable
  // from the Swift client API, depending on the remote server
  final FileStatus[] listOfFileBlocks = store.listSubPaths(file.getPath(),
                                                           false,
                                                           true);
  List<URI> locations = new ArrayList<URI>();
  if (listOfFileBlocks.length > 1) {
    for (FileStatus fileStatus : listOfFileBlocks) {
      if (SwiftObjectPath.fromPath(uri, fileStatus.getPath())
              .equals(SwiftObjectPath.fromPath(uri, file.getPath()))) {
        continue;
      }
      locations.addAll(store.getObjectLocation(fileStatus.getPath()));
    }
  } else {
    locations = store.getObjectLocation(file.getPath());
  }

  if (locations.isEmpty()) {
    LOG.debug("No locations returned for " + file.getPath());
    //no locations were returned for the object
    //fall back to the superclass

    String[] name = {SwiftProtocolConstants.BLOCK_LOCATION};
    String[] host = { "localhost" };
    String[] topology={SwiftProtocolConstants.TOPOLOGY_PATH};
    return new BlockLocation[] {
      new BlockLocation(name, host, topology,0, file.getLen())
    };
  }

  final String[] names = new String[locations.size()];
  final String[] hosts = new String[locations.size()];
  int i = 0;
  for (URI location : locations) {
    hosts[i] = location.getHost();
    names[i] = location.getAuthority();
    i++;
  }
  return new BlockLocation[]{
          new BlockLocation(names, hosts, 0, file.getLen())
  };
}
 

@Test(timeout = SWIFT_BULK_IO_TEST_TIMEOUT)
public void testPartitionPropertyPropagatesToConf() throws Throwable {
  assertEquals(1,
               getConf().getInt(SwiftProtocolConstants.SWIFT_PARTITION_SIZE,
                                0));
}
 

@Override
protected Configuration createConfiguration() {
  Configuration conf = super.createConfiguration();
  conf.set(SwiftProtocolConstants.SWIFT_REQUEST_SIZE, "1");
  return conf;
}
 

/**
 * Return an array containing hostnames, offset and size of
 * portions of the given file.  For a nonexistent
 * file or regions, null will be returned.
 * <p>
 * This call is most helpful with DFS, where it returns
 * hostnames of machines that contain the given file.
 * <p>
 * The FileSystem will simply return an elt containing 'localhost'.
 */
@Override
public BlockLocation[] getFileBlockLocations(FileStatus file,
                                             long start,
                                             long len) throws IOException {
  //argument checks
  if (file == null) {
    return null;
  }

  if (start < 0 || len < 0) {
    throw new IllegalArgumentException("Negative start or len parameter" +
                                       " to getFileBlockLocations");
  }
  if (file.getLen() <= start) {
    return new BlockLocation[0];
  }

  // Check if requested file in Swift is more than 5Gb. In this case
  // each block has its own location -which may be determinable
  // from the Swift client API, depending on the remote server
  final FileStatus[] listOfFileBlocks = store.listSubPaths(file.getPath(),
                                                           false,
                                                           true);
  List<URI> locations = new ArrayList<URI>();
  if (listOfFileBlocks.length > 1) {
    for (FileStatus fileStatus : listOfFileBlocks) {
      if (SwiftObjectPath.fromPath(uri, fileStatus.getPath())
              .equals(SwiftObjectPath.fromPath(uri, file.getPath()))) {
        continue;
      }
      locations.addAll(store.getObjectLocation(fileStatus.getPath()));
    }
  } else {
    locations = store.getObjectLocation(file.getPath());
  }

  if (locations.isEmpty()) {
    LOG.debug("No locations returned for " + file.getPath());
    //no locations were returned for the object
    //fall back to the superclass

    String[] name = {SwiftProtocolConstants.BLOCK_LOCATION};
    String[] host = { "localhost" };
    String[] topology={SwiftProtocolConstants.TOPOLOGY_PATH};
    return new BlockLocation[] {
      new BlockLocation(name, host, topology,0, file.getLen())
    };
  }

  final String[] names = new String[locations.size()];
  final String[] hosts = new String[locations.size()];
  int i = 0;
  for (URI location : locations) {
    hosts[i] = location.getHost();
    names[i] = location.getAuthority();
    i++;
  }
  return new BlockLocation[]{
          new BlockLocation(names, hosts, 0, file.getLen())
  };
}
 

@Test(timeout = SWIFT_BULK_IO_TEST_TIMEOUT)
public void testPartitionPropertyPropagatesToConf() throws Throwable {
  assertEquals(1,
               getConf().getInt(SwiftProtocolConstants.SWIFT_PARTITION_SIZE,
                                0));
}
 

@Override
protected Configuration createConfiguration() {
  Configuration conf = super.createConfiguration();
  conf.set(SwiftProtocolConstants.SWIFT_REQUEST_SIZE, "1");
  return conf;
}
 

/**
 * Return an array containing hostnames, offset and size of
 * portions of the given file.  For a nonexistent
 * file or regions, null will be returned.
 * <p/>
 * This call is most helpful with DFS, where it returns
 * hostnames of machines that contain the given file.
 * <p/>
 * The FileSystem will simply return an elt containing 'localhost'.
 */
@Override
public BlockLocation[] getFileBlockLocations(FileStatus file,
                                             long start,
                                             long len) throws IOException {
  //argument checks
  if (file == null) {
    return null;
  }
  if (file.isDir()) {
    return new BlockLocation[0];
  }

  if (start < 0 || len < 0) {
    throw new IllegalArgumentException("Negative start or len parameter" +
                                       " to getFileBlockLocations");
  }
  if (file.getLen() <= start) {
    return new BlockLocation[0];
  }

  // Check if requested file in Swift is more than 5Gb. In this case
  // each block has its own location -which may be determinable
  // from the Swift client API, depending on the remote server
  final FileStatus[] listOfFileBlocks;
  if (file instanceof SwiftFileStatus && ((SwiftFileStatus)file).isDLO()) {
    listOfFileBlocks = store.listSegments(file, true);
  } else {
    listOfFileBlocks = null;
  }

  List<URI> locations = new ArrayList<URI>();
  if (listOfFileBlocks != null && listOfFileBlocks.length > 1) {
    for (FileStatus fileStatus : listOfFileBlocks) {
      if (SwiftObjectPath.fromPath(uri, fileStatus.getPath())
              .equals(SwiftObjectPath.fromPath(uri, file.getPath()))) {
        continue;
      }
      locations.addAll(store.getObjectLocation(fileStatus.getPath()));
    }
  } else {
    locations = store.getObjectLocation(file.getPath());
  }

  if (locations.isEmpty()) {
    LOG.debug("No locations returned for " + file.getPath());
    //no locations were returned for the object
    //fall back to the superclass

    String[] name = {SwiftProtocolConstants.BLOCK_LOCATION};
    String[] host = { "localhost" };
    String[] topology={SwiftProtocolConstants.TOPOLOGY_PATH};
    return new BlockLocation[] {
      new BlockLocation(name, host, topology,0, file.getLen())
    };
  }

  final String[] names = new String[locations.size()];
  final String[] hosts = new String[locations.size()];
  int i = 0;
  for (URI location : locations) {
    hosts[i] = location.getHost();
    names[i] = location.getAuthority();
    i++;
  }
  return new BlockLocation[]{
          new BlockLocation(names, hosts, 0, file.getLen())
  };
}
 

@Test(timeout = SWIFT_BULK_IO_TEST_TIMEOUT)
public void testPartitionPropertyPropagatesToConf() throws Throwable {
  assertEquals(1,
               getConf().getInt(SwiftProtocolConstants.SWIFT_PARTITION_SIZE,
                                0));
}
 

@Override
protected Configuration createConfiguration() {
  Configuration conf = super.createConfiguration();
  conf.set(SwiftProtocolConstants.SWIFT_REQUEST_SIZE, "1");
  return conf;
}