import org.javaswift.joss.model.StoredObject; //导入依赖的package包/类
@Override
public void flush(Map<TopicPartition, OffsetAndMetadata> currentOffsets) {
  try {
    Iterator<BulkedData> items = bulker.convert();

    while (items.hasNext()) {
      final BulkedData item = items.next();

      final Container container = this.account.getContainer(item.getContainer());
      if (!container.exists()) {
        container.create();
      }

      final StoredObject object = container.getObject(item.getObject());
      object.uploadObject(item.getStream());
    }
  } catch (Exception exception) {
    log.error("Failed to add record", exception);

    throw new ConnectException("Failed to flush", exception);
  }
}
 

import org.javaswift.joss.model.StoredObject; //导入依赖的package包/类
@Override
public boolean delete(String hostName, Path path, boolean recursive) throws IOException {
  String obj = path.toString();
  if (path.toString().startsWith(hostName)) {
    obj = getObjName(hostName, path);
  }
  LOG.debug("Object name to delete {}. Path {}", obj, path.toString());
  try {
    StoredObject so = mJossAccount.getAccount().getContainer(container)
        .getObject(obj);
    if (so.exists()) {
      so.delete();
      objectCache.remove(obj);
    }
  } catch (Exception e) {
    LOG.warn(e.getMessage());
    LOG.warn("Delete on {} resulted in FileNotFound exception", path);
    return false;
  }
  return true;
}
 

import org.javaswift.joss.model.StoredObject; //导入依赖的package包/类
/**
 * Checks if container/object exists and verifies
 * that it contains Data-Origin=stocator metadata
 * If so, object was created by Spark.
 *
 * @param objectName
 * @return boolean if object was created by Spark
 */
private boolean isSparkOrigin(String objectName) {
  LOG.trace("Check if created by Stocator: {}", objectName);
  if (cachedSparkOriginated.containsKey(objectName)) {
    return cachedSparkOriginated.get(objectName).booleanValue();
  }
  String obj = objectName;
  Boolean sparkOriginated = Boolean.FALSE;
  StoredObject so = mJossAccount.getAccount().getContainer(container).getObject(obj);
  if (so != null && so.exists()) {
    Object sparkOrigin = so.getMetadata("Data-Origin");
    if (sparkOrigin != null) {
      String tmp = (String) sparkOrigin;
      if (tmp.equals("stocator")) {
        sparkOriginated = Boolean.TRUE;
        LOG.trace("Object {} was created by Stocator", objectName);
      }
    }
  }
  cachedSparkOriginated.put(objectName, sparkOriginated);
  return sparkOriginated.booleanValue();
}
 

import org.javaswift.joss.model.StoredObject; //导入依赖的package包/类
/**
 * Checks if container/object contains
 * container/object/_SUCCESS
 * If so, this object was created by successful Hadoop job
 *
 * @param objectName
 * @return boolean if job is successful
 */
private boolean isJobSuccessful(String objectName) {
  LOG.trace("Checking if job completed successfull for {}", objectName);
  if (cachedSparkJobsStatus.containsKey(objectName)) {
    return cachedSparkJobsStatus.get(objectName).booleanValue();
  }
  String obj = objectName;
  Account account = mJossAccount.getAccount();
  LOG.trace("HEAD {}", obj + "/" + HADOOP_SUCCESS);
  StoredObject so = account.getContainer(container).getObject(obj
      + "/" + HADOOP_SUCCESS);
  Boolean isJobOK = Boolean.FALSE;
  if (so.exists()) {
    LOG.debug("{} exists", obj + "/" + HADOOP_SUCCESS);
    isJobOK = Boolean.TRUE;
  }
  cachedSparkJobsStatus.put(objectName, isJobOK);
  return isJobOK.booleanValue();
}
 

import org.javaswift.joss.model.StoredObject; //导入依赖的package包/类
@Override
public boolean rename(String hostName, String srcPath, String dstPath) throws IOException {
  LOG.debug("Rename from {} to {}. hostname is {}", srcPath, dstPath, hostName);
  String objNameSrc = srcPath.toString();
  if (srcPath.toString().startsWith(hostName)) {
    objNameSrc = getObjName(hostName, srcPath);
  }
  String objNameDst = dstPath.toString();
  if (objNameDst.toString().startsWith(hostName)) {
    objNameDst = getObjName(hostName, dstPath);
  }
  if (stocatorPath.isTemporaryPathContain(objNameSrc)) {
    LOG.debug("Rename on the temp object {}. Return true", objNameSrc);
    return true;
  }
  LOG.debug("Rename modified from {} to {}", objNameSrc, objNameDst);
  Container cont = mJossAccount.getAccount().getContainer(container);
  StoredObject so = cont.getObject(objNameSrc);
  StoredObject soDst = cont.getObject(objNameDst);
  so.copyObject(cont, soDst);
  return true;
}
 

import org.javaswift.joss.model.StoredObject; //导入依赖的package包/类
/**
 * The get function will first search for the object in the cache.
 * If not found will issue a HEAD request for the object metadata
 * and add the object to the cache.
 *
 * @param objName object name
 * @return cached entry of the object
 * @throws IOException if failed to parse time stamp
 */
public SwiftCachedObject get(String objName) throws IOException {
  LOG.trace("Get from cache  {} ", objName);
  SwiftCachedObject res = cache.get(objName);
  if (res == null) {
    LOG.trace("Cache get:  {} is not in the cache. Access Swift to get content length", objName);
    StoredObject rawObj = container.getObject(removeTrailingSlash(objName));
    if (rawObj != null && rawObj.exists()) {
      res = new SwiftCachedObject(rawObj.getContentLength(),
        Utils.lastModifiedAsLong(rawObj.getLastModified()));
      put(objName, res);
    } else {
      return null;
    }
  }
  return res;
}
 

import org.javaswift.joss.model.StoredObject; //导入依赖的package包/类
@Override
public void sendNextPart(MultiPartUpload mpu) {
    Integer currentComponent = (Integer) mpu.getObject("currentChunk");
    Integer numChunks = (Integer) mpu.getObject("numOfChunks");
    String localPath = mpu.getObject("localPath").toString();
    
    if(currentComponent < numChunks)
    {            
        String fileName = mpu.getObject("fileName").toString();
        fileName = fileName + currentComponent;
        
        StoredObject object = segmentContainer.getObject(fileName);
        object.uploadObject(new File(localPath + fileName));            
        mpu.putObject("currentChunk", ++currentComponent);
    }
    else
    {
        mpu.setFinished();
    }
}
 

import org.javaswift.joss.model.StoredObject; //导入依赖的package包/类
public RemoteItem (StoredObject storedObject) throws IOException
{
	super () ;
	if (storedObject == null)
		throw new IllegalArgumentException ("storedObject cannot be null") ;
	this.storedObject = storedObject ;
	this.exists = storedObject.exists() ;
	this.size = (exists) ? (storedObject.getContentLength()) : (-2) ;
	if (exists)
	{
		// For segmented object, the etag may be bounded by "".
		String etag = storedObject.getEtag()  ;
		if (etag.startsWith("\"")) 
			etag = etag.replace("\"", "") ;
		this.md5 = etag ;
	}
	else
		this.md5 = "" ;
}
 

import org.javaswift.joss.model.StoredObject; //导入依赖的package包/类
private void uploadAndOverwrite (boolean overwrite) throws IOException
{
	final String objName = "testObject.dat" ;
	
    Container container = account.getContainer("x").create();
    StoredObject object = container.getObject(objName);
    object.uploadObject(TestUtils.getTestFile(tmpFolder, objName, 4096));

    assertTrue (object.exists()) ;
    String md5 = object.getEtag() ;
    
    // upload a new file with the same name, the overwrite flag is set to overwrite
    ops.uploadFiles(container, null, new File [] {TestUtils.getTestFile(tmpFolder, objName, 4096)}, overwrite, stopRequester, callback);
    
    StoredObject newObject = container.getObject(objName);
    assertTrue (newObject.exists()) ;
    
    // we assume no collision ;-)
    assertTrue (md5.equals(newObject.getEtag()) == !overwrite) ;
}
 

import org.javaswift.joss.model.StoredObject; //导入依赖的package包/类
public static void CreateObject(String username,String passwod,String dataNode, String container, String key, InputStream content)throws Exception
{
	try{
		Account client = ViPRSwiftClient(username, passwod, dataNode);
		Container myContainer = client.getContainer(container);
		if (!myContainer.exists()) {
			myContainer.create();
			myContainer.makePublic();
		}
		StoredObject object = myContainer.getObject(key);
		object.uploadObject(content);

	}
	catch(Exception e)
	{
		System.out.println("error during creating "+key);
		e.printStackTrace();
	}

}
 

import org.javaswift.joss.model.StoredObject; //导入依赖的package包/类
@Override
  public String getSumOfSegmentsMd5 (StoredObject obj)
  {
  	if (obj == null)
  		return FileUtils.emptyMd5 ;
  	List<StoredObject> segments = getSegmentsList (obj) ;
  	StringBuilder sb = new StringBuilder () ;
  	for (StoredObject so : segments){
  		sb.append(so.getEtag()) ;
  	}
  	InputStream stream = new java.io.ByteArrayInputStream (sb.toString().getBytes(StandardCharsets.UTF_8));
  	try 
  	{
	return FileUtils.readAllAndgetMD5(stream) ;
} 
  	catch (IOException e) {
	logger.error("Error occurred while computing md5 value", e) ;
}
  	return "" ;
  }
 

import org.javaswift.joss.model.StoredObject; //导入依赖的package包/类
@Override
public long getActualSegmentSize (StoredObject obj)
{
	final long notFound = -1 ;
	Container segCont = getSegmentsContainer (obj, false) ;
	if (segCont == null || !segCont.exists())
		return notFound ;
	StoredObject segObj = getObjectSegment ((AbstractContainer)segCont, (AbstractStoredObject)obj, Long.valueOf(1)) ;
	if (segObj == null || !segObj.exists())
	{
		// the object may have been segmented using another convention,
		// we check using the manifest
 	List<StoredObject> sgl = getSegmentsList(obj, 0, 1) ;
 	if (!sgl.isEmpty())
 		segObj = sgl.get(0) ;
	}
	if (segObj == null || !segObj.exists())
		return notFound ;
	return segObj.getContentLength() ;
}
 

import org.javaswift.joss.model.StoredObject; //导入依赖的package包/类
public DifferencesManagementPanel (Container container, StoredObject srcStoredObject, File srcFile, Collection<Pair<? extends ComparisonItem, ? extends ComparisonItem> >  discrepancies, SwiftOperations ops, SwiftCallback callback, HasLocalizedStrings stringsBundle, JDialog owner, Dimension parentDim)
{
	super (new BorderLayout(0, 0)) ;
	if (discrepancies == null)
		throw new IllegalArgumentException () ;
	
	setBorder(BorderFactory.createEmptyBorder(4, 4, 4, 4));
	
	this.owner = owner ;
	this.discrepancies = discrepancies ;
	this.stringsBundle = stringsBundle ;
	this.container = container ;
	this.srcStoredObject = srcStoredObject ;
	this.srcFile = srcFile ;
	this.ops = ops ;
	this.callback = callback ;

	if (parentDim != null)
	{
		this.colWidth = (int) Math.max(colWidth, (0.8 * (parentDim.getWidth())) / 3.0 - 20) ;
		this.listHeight = (int) Math.min(listHeight, parentDim.getHeight() - 50) ;
	}
	
	initPanel () ;
}
 

import org.javaswift.joss.model.StoredObject; //导入依赖的package包/类
private File getDestinationFile (StoredObject srcDirRootStoredObject, File destDirRoot, StoredObject srcStoredObject)
{
	StringBuilder pathBuilder = new StringBuilder () ;
	pathBuilder.append(destDirRoot.getPath ()) ;
	if (!destDirRoot.getPath ().endsWith(File.separator))
		pathBuilder.append(File.separator) ;
	String dirName = srcDirRootStoredObject.getName() ;
	int index = dirName.lastIndexOf(SwiftUtils.separator) ;
	if (index >= 0)
		dirName = dirName.substring(index + 1) ;
	pathBuilder.append(dirName) ;
	pathBuilder.append(srcStoredObject.getName().replaceFirst(srcDirRootStoredObject.getName(),"").trim()) ;
	String path = pathBuilder.toString() ;
	if (!SwiftUtils.separator.equals(File.separator))
		path = path.replace(SwiftUtils.separator, File.separator) ;
	
	return new File(path);
}
 

import org.javaswift.joss.model.StoredObject; //导入依赖的package包/类
@Test
public void shouldReturnFileObject () throws IOException
{
	Container container = account.getContainer("x").create() ;
	
	final String root = "root" ;
	
	StoredObject rootObj = container.getObject(root) ;
	rootObj.uploadObject(new UploadInstructions (new byte[] {}).setContentType(SwiftUtils.directoryContentType)) ;
	
	File rootDirectory = TestUtils.getTestDirectoryWithFiles (tmpFolder, root, "file", 2) ;
	Path file = Paths.get(new File (rootDirectory.getPath() + File.separator + "file_1").getPath()) ;
	assertTrue (Files.exists(file)) ;

	DifferencesFinder.LocalItem li = new DifferencesFinder.LocalItem(file, Paths.get(rootDirectory.getPath()), Long.MAX_VALUE) ;
	StoredObject obj = DifferencesFinder.getObject(container, li, rootObj) ;
	
	assertTrue (obj.getBareName().equals(file.getFileName().toString())) ;
	assertTrue (obj.getName().equals(root + SwiftUtils.separator + file.getFileName().toString())) ;
}
 

import org.javaswift.joss.model.StoredObject; //导入依赖的package包/类
/**
 * {@inheritDoc}.
 */
@Override
public synchronized void emptyContainer(Container container, StopRequester stopRequester, SwiftCallback callback) {
	
	CheckAccount () ;
	
	try
	{
     for (StoredObject so : eagerFetchStoredObjects(container)) 
     {
     	if (!keepGoing (stopRequester, callback))
     		return ;
     	
     	if (so.exists())
     		so.delete();
         callback.onNumberOfCalls(account.getNumberOfCalls());
     }
     logger.info(String.format("Container %s has been emptied", container.getName()));
	}
	finally
	{
     reloadContainer(container, callback);
     callback.onNumberOfCalls(account.getNumberOfCalls());
	}
    
}
 

import org.javaswift.joss.model.StoredObject; //导入依赖的package包/类
/**
 * {@inheritDoc}.
 */
@Override
public synchronized void purgeContainer(Container container, StopRequester stopRequester, SwiftCallback callback) {
	
	CheckAccount () ;
	
	try
	{
     for (StoredObject so : eagerFetchStoredObjects(container)) 
     {
     	if (!keepGoing (stopRequester, callback))
     		return ;
     	
     	if (so.exists())
     		so.delete();
         callback.onNumberOfCalls(account.getNumberOfCalls());
     }
     container.delete();
     logger.info(String.format("Container %s has been removed", container.getName()));
	}
	finally
	{
     callback.onUpdateContainers(eagerFetchContainers(account));
     callback.onNumberOfCalls(account.getNumberOfCalls());
	}
}
 

import org.javaswift.joss.model.StoredObject; //导入依赖的package包/类
private void addedObjectToContainer(Container container, List<StoredObject> newObjects, SwiftCallback callback) {
  	
  	if (container == null)
  		throw new AssertionError ("container cannot be null") ;
  	
  	container.reload();

  	if (newObjects == null || newObjects.isEmpty())
  		return ;
  	
int page = 0;
while (page * MAX_PAGE_SIZE < newObjects.size())
{
	callback.onAppendStoredObjects(container, page, newObjects.subList(page * MAX_PAGE_SIZE, Math.min(newObjects.size(), (page + 1) * MAX_PAGE_SIZE))) ;
	++page ;
}
  }
 

import org.javaswift.joss.model.StoredObject; //导入依赖的package包/类
@Test
  public void shouldFindOutWhetherItIsSegmented() 
  {
  	uploadAndCheckSegmentedFile () ;
  	
  	StoredObject object = container.getObject(objName);
  	assertTrue(largeObjectManager.isSegmented(object)) ;
  	
  	// upload a non-segmented file...
      try 
      {
	File smallFile = TestUtils.getTestFile (tmpFolder, "nonSegmentedData.dat", segmentSize / 2) ;
	StoredObject obj = container.getObject(smallFile.getName());
	obj.uploadObject(smallFile);
      	assertTrue (obj.exists()) ;
      	
      	assertFalse(largeObjectManager.isSegmented(obj)) ;
} 
      catch (IOException e) 
{
      	logger.error ("Error occurred in shouldFindOutThatItIsSegmented", e) ;
      	assertFalse(true) ;
}
  }
 

import org.javaswift.joss.model.StoredObject; //导入依赖的package包/类
@SuppressWarnings("unused")
  private void onDownloadRemote ()
  {
  	
  	List<ComparisonItem> listItems = getSelectedRemoteItems (true) ;
  	List<Pair<? extends StoredObject, ? extends File> > pairParentObjectFiles = new ArrayList<> () ;
  	for (ComparisonItem ri : listItems)
  	{
  		if (ri == null)
  			continue ;
  		if (!(ri instanceof DifferencesFinder.RemoteItem))
  			continue ;
  		pairParentObjectFiles.add(Pair.newPair(DifferencesFinder.getObject(container, ri, srcStoredObject), DifferencesFinder.getFile(ri, srcFile))) ;
  	}
  	try 
  	{
	ops.downloadStoredObject(container, pairParentObjectFiles, false, null, callback);
} 
  	catch (IOException e) 
  	{
  		logger.error("Error occurred while downloading files.", e);
}	 	
  }