diff --git a/datastore_server/source/java/ch/systemsx/cisd/openbis/dss/generic/shared/content/ContentCache.java b/datastore_server/source/java/ch/systemsx/cisd/openbis/dss/generic/shared/content/ContentCache.java
index ac9ef4e6597c5e637af14fbf285afccc0b10bedd..eeed7b713e43fb74431a8af1e9bae9c070a5b8a0 100644
--- a/datastore_server/source/java/ch/systemsx/cisd/openbis/dss/generic/shared/content/ContentCache.java
+++ b/datastore_server/source/java/ch/systemsx/cisd/openbis/dss/generic/shared/content/ContentCache.java
@@ -296,12 +296,19 @@ public class ContentCache implements IContentCache, InitializingBean
return new InputStream()
{
private boolean closed;
+ private boolean eof;
@Override
public int read() throws IOException
{
int b = inputStream.read();
- fileOutputStream.write(b);
+ if (b < 0)
+ {
+ eof = true;
+ } else
+ {
+ fileOutputStream.write(b);
+ }
return b;
}
@@ -312,6 +319,10 @@ public class ContentCache implements IContentCache, InitializingBean
if (count >= 0)
{
fileOutputStream.write(b, off, count);
+ eof = count < len;
+ } else
+ {
+ eof = true;
}
return count;
}
@@ -325,9 +336,12 @@ public class ContentCache implements IContentCache, InitializingBean
}
inputStream.close();
fileOutputStream.close();
- moveDownloadedFileToCache(tempFile, pathInWorkspace,
- dataSetLocation.getDataSetCode());
- persistenceManager.requestPersistence();
+ if (eof)
+ {
+ moveDownloadedFileToCache(tempFile, pathInWorkspace,
+ dataSetLocation.getDataSetCode());
+ persistenceManager.requestPersistence();
+ }
closed = true;
fileLockManager.unlock(pathInWorkspace);
}
@@ -351,8 +365,8 @@ public class ContentCache implements IContentCache, InitializingBean
try
{
input = createInputStream(sessionToken, dataSetLocation, path);
- File downloadedFile = createFileFromInputStream(dataSetLocation, path, input);
String pathInWorkspace = createPathInWorkspace(CACHE_FOLDER, dataSetLocation, path);
+ File downloadedFile = createFileFromInputStream(input);
moveDownloadedFileToCache(downloadedFile, pathInWorkspace,
dataSetLocation.getDataSetCode());
} catch (Exception ex)
@@ -461,8 +475,7 @@ public class ContentCache implements IContentCache, InitializingBean
}
}
- private File createFileFromInputStream(IDatasetLocation dataSetLocation, DataSetPathInfo path,
- InputStream inputStream)
+ private File createFileFromInputStream(InputStream inputStream)
{
File file = createTempFile();
OutputStream ostream = null;
@@ -483,10 +496,16 @@ public class ContentCache implements IContentCache, InitializingBean
private File createTempFile()
{
- String relativePath = DOWNLOADING_FOLDER + "/" + Thread.currentThread().getId();
- File file = new File(workspace, relativePath);
- createFolder(file.getParentFile());
- return file;
+ File downLoadingFolder = new File(workspace, DOWNLOADING_FOLDER);
+ createFolder(downLoadingFolder);
+ try
+ {
+ File file = File.createTempFile("file-", null, downLoadingFolder);
+ return file;
+ } catch (IOException ex)
+ {
+ throw CheckedExceptionTunnel.wrapIfNecessary(ex);
+ }
}
private void createFolder(File folder)
@@ -551,14 +570,24 @@ public class ContentCache implements IContentCache, InitializingBean
String dataSet = entry.getKey();
if (isDataSetLocked(dataSet) == false)
{
- fileOperations.removeRecursivelyQueueing(new File(workspace, dataSet));
- dataSetInfos.remove(dataSet);
- totalSize -= info.size;
- operationLog.info("Cached files for data set " + dataSet
- + " have been removed.");
- if (totalSize < maxWorkspaceSize)
+ File fileToRemove = new File(workspace, dataSet);
+ boolean success = fileOperations.removeRecursivelyQueueing(fileToRemove);
+ if (success)
+ {
+ synchronized (dataSetInfos)
+ {
+ dataSetInfos.remove(dataSet);
+ }
+ totalSize -= info.size;
+ operationLog.info("Cached files for data set " + dataSet
+ + " have been removed.");
+ if (totalSize < maxWorkspaceSize)
+ {
+ break;
+ }
+ } else
{
- break;
+ operationLog.error("Couldn't remove " + fileToRemove);
}
}
}
diff --git a/datastore_server/sourceTest/java/ch/systemsx/cisd/openbis/dss/generic/shared/content/AbstractRemoteHierarchicalContentTestCase.java b/datastore_server/sourceTest/java/ch/systemsx/cisd/openbis/dss/generic/shared/content/AbstractRemoteHierarchicalContentTestCase.java
index b012e7689d7f52fbc270e1ca6fdc2d67f68a0023..58d8ca10acbbec10bab1de0b014c25e00f5849c8 100644
--- a/datastore_server/sourceTest/java/ch/systemsx/cisd/openbis/dss/generic/shared/content/AbstractRemoteHierarchicalContentTestCase.java
+++ b/datastore_server/sourceTest/java/ch/systemsx/cisd/openbis/dss/generic/shared/content/AbstractRemoteHierarchicalContentTestCase.java
@@ -83,7 +83,7 @@ public abstract class AbstractRemoteHierarchicalContentTestCase extends Abstract
protected IPersistenceManager persistenceManager;
- private HashMap<String, DataSetInfo> dataSetInfos;
+ protected HashMap<String, DataSetInfo> dataSetInfos;
@BeforeMethod
public void setUpBasicFixture()
@@ -126,14 +126,6 @@ public abstract class AbstractRemoteHierarchicalContentTestCase extends Abstract
context.assertIsSatisfied();
}
- protected void assertDataSetInfos(String dataSetCode, long expectedSize,
- long expectedLastModified)
- {
- DataSetInfo dataSetInfo = dataSetInfos.get(dataSetCode);
- assertEquals(expectedLastModified, dataSetInfo.lastModified);
- assertEquals(expectedSize, dataSetInfo.size);
- }
-
protected ContentCache createCache()
{
context.checking(new Expectations()
diff --git a/datastore_server/sourceTest/java/ch/systemsx/cisd/openbis/dss/generic/shared/content/ContentCacheTest.java b/datastore_server/sourceTest/java/ch/systemsx/cisd/openbis/dss/generic/shared/content/ContentCacheTest.java
index dc89bcdbcbf918102b16719aea05f3c6469c6356..de8d0930d8e56ccb601c1a9455344d414906dab5 100644
--- a/datastore_server/sourceTest/java/ch/systemsx/cisd/openbis/dss/generic/shared/content/ContentCacheTest.java
+++ b/datastore_server/sourceTest/java/ch/systemsx/cisd/openbis/dss/generic/shared/content/ContentCacheTest.java
@@ -27,11 +27,13 @@ import org.jmock.Expectations;
import org.testng.annotations.Test;
import ch.systemsx.cisd.base.exceptions.CheckedExceptionTunnel;
+import ch.systemsx.cisd.base.utilities.OSUtilities;
import ch.systemsx.cisd.common.concurrent.MessageChannel;
import ch.systemsx.cisd.common.concurrent.MessageChannelBuilder;
import ch.systemsx.cisd.common.filesystem.FileUtilities;
import ch.systemsx.cisd.common.logging.ConsoleLogger;
import ch.systemsx.cisd.common.test.ProxyAction;
+import ch.systemsx.cisd.openbis.dss.generic.shared.content.ContentCache.DataSetInfo;
import ch.systemsx.cisd.openbis.dss.generic.shared.dto.DataSetPathInfo;
/**
@@ -94,6 +96,7 @@ public class ContentCacheTest extends AbstractRemoteHierarchicalContentTestCase
File file = cache.getFile(SESSION_TOKEN, DATA_SET_LOCATION, pathInfo);
assertEquals(FILE1_CONTENT, FileUtilities.loadToString(file).trim());
+ assertDataSetInfos(DATA_SET_CODE, 1, 1, 1000);
context.assertIsSatisfied();
}
@@ -110,7 +113,54 @@ public class ContentCacheTest extends AbstractRemoteHierarchicalContentTestCase
assertDataSetInfos(DATA_SET_CODE, FILE1_CONTENT.length(), 1000);
context.assertIsSatisfied();
}
+
+ @Test
+ public void testGetInputStreamForFileNotInCacheReadingBytePerByte() throws IOException
+ {
+ final DataSetPathInfo pathInfo1 = prepareForDownloading(remoteFile1);
+ prepareRequestPersistence(1);
+
+ ContentCache cache = createCache();
+ InputStream inputStream = cache.getInputStream(SESSION_TOKEN, DATA_SET_LOCATION, pathInfo1);
+
+ StringBuilder builder = new StringBuilder();
+ while (true)
+ {
+ int b = inputStream.read();
+ if (b < 0)
+ {
+ break;
+ }
+ builder.append((char) b);
+ }
+ inputStream.close();
+
+ assertEquals(FILE1_CONTENT, builder.toString());
+ assertDataSetInfos(DATA_SET_CODE, FILE1_CONTENT.length(), 1000);
+ context.assertIsSatisfied();
+ }
+ @Test
+ public void testGetInputStreamForFileNotInCacheAndInterruptReading() throws IOException
+ {
+ prepareForDownloading(remoteFile1);
+ final DataSetPathInfo pathInfo1 = prepareForDownloading(remoteFile1);
+ prepareRequestPersistence(1);
+
+ ContentCache cache = createCache();
+ InputStream inputStream = cache.getInputStream(SESSION_TOKEN, DATA_SET_LOCATION, pathInfo1);
+
+ byte[] bytes = new byte[100];
+ assertEquals(11, inputStream.read(bytes, 0, 11));
+ assertEquals(FILE1_CONTENT.substring(0, 11), new String(bytes, 0, 11));
+ inputStream.close();
+
+ inputStream = cache.getInputStream(SESSION_TOKEN, DATA_SET_LOCATION, pathInfo1);
+ assertEquals(FILE1_CONTENT, readContent(inputStream, true));
+ assertDataSetInfos(DATA_SET_CODE, FILE1_CONTENT.length(), 1000);
+ context.assertIsSatisfied();
+ }
+
@Test
public void testGetInputStreamForFileInCache()
{
@@ -127,9 +177,46 @@ public class ContentCacheTest extends AbstractRemoteHierarchicalContentTestCase
InputStream inputStream = cache.getInputStream(SESSION_TOKEN, DATA_SET_LOCATION, pathInfo);
assertEquals(FILE1_CONTENT, readContent(inputStream, true));
+ assertDataSetInfos(DATA_SET_CODE, 1, 1, 1000);
context.assertIsSatisfied();
}
+ @Test
+ public void testGetInputStreamsForTwoFilesNotInCacheAndReadThemAtTheSameTime() throws IOException
+ {
+ final DataSetPathInfo pathInfo1 = prepareForDownloading(remoteFile1);
+ final DataSetPathInfo pathInfo2 = prepareForDownloading(remoteFile2);
+ prepareRequestPersistence(4);
+
+ ContentCache cache = createCache();
+ InputStream inputStream1 =
+ cache.getInputStream(SESSION_TOKEN, DATA_SET_LOCATION, pathInfo1);
+ InputStream inputStream2 =
+ cache.getInputStream(SESSION_TOKEN, DATA_SET_LOCATION, pathInfo2);
+
+ byte[] bytes1 = new byte[100];
+ byte[] bytes2 = new byte[100];
+ assertEquals(11, inputStream1.read(bytes1, 0, 11));
+ assertEquals(11, inputStream2.read(bytes2, 0, 11));
+ assertEquals(3, inputStream1.read(bytes1, 11, 100 - 11));
+ assertEquals(3, inputStream2.read(bytes2, 11, 100 - 11));
+ inputStream1.close();
+ inputStream2.close();
+
+ assertEquals(FILE1_CONTENT, new String(bytes1, 0, FILE1_CONTENT.length()));
+ assertEquals(FILE2_CONTENT, new String(bytes2, 0, FILE2_CONTENT.length()));
+ assertEquals(
+ FILE1_CONTENT,
+ FileUtilities.loadToString(
+ cache.getFile(SESSION_TOKEN, DATA_SET_LOCATION, pathInfo1)).trim());
+ assertEquals(
+ FILE2_CONTENT,
+ FileUtilities.loadToString(
+ cache.getFile(SESSION_TOKEN, DATA_SET_LOCATION, pathInfo2)).trim());
+
+ context.assertIsSatisfied();
+ }
+
@Test(invocationCount = 1, invocationTimeOut = 10000)
public void testGetInputStreamForSameContentInTwoThreads()
{
@@ -323,6 +410,25 @@ public class ContentCacheTest extends AbstractRemoteHierarchicalContentTestCase
context.assertIsSatisfied();
}
+ private void assertDataSetInfos(String dataSetCode, int expectedNumberOfSmallFiles, int expectedNumberOfFolders,
+ long expectedLastModified)
+ {
+ long expectedSize = expectedNumberOfSmallFiles;
+ if (OSUtilities.isMacOS() == false)
+ {
+ expectedSize += expectedNumberOfFolders;
+ }
+ assertDataSetInfos(dataSetCode, expectedSize * 4096, expectedLastModified);
+ }
+
+ private void assertDataSetInfos(String dataSetCode, long expectedSize,
+ long expectedLastModified)
+ {
+ DataSetInfo dataSetInfo = dataSetInfos.get(dataSetCode);
+ assertEquals(expectedLastModified, dataSetInfo.lastModified);
+ assertEquals(expectedSize, dataSetInfo.size);
+ }
+
private String readContent(InputStream inputStream, boolean closeStream)
{
try