May 2010 - Diswork-commits - mailman.cloud.codelutin.com

r23 - in trunk/diswork-fs/src: main/java/org/nuiton/disworkfs main/java/org/nuiton/disworkfs/services main/java/org/nuiton/disworkfs/split main/java/org/nuiton/disworkfs/transport main/java/org/nuiton/disworkfs/transport/jgroups main/resources test/java/org/nuiton/disworkfs test/java/org/nuiton/disworkfs/split
by bleny＠users.nuiton.org May 6, 2010

May 6, 2010

Author: bleny Date: 2010-05-06 10:25:12 +0200 (Thu, 06 May 2010) New Revision: 23 Url: http://nuiton.org/repositories/revision/diswork/23 Log: logs, lecture/ecriture des fichiers sur le FS, menage Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkConfig.java Removed: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/config/ Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkFileSystem.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkServicesManager.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/RunMe.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/SimpleDownload.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/AbstractDisworkService.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/DownloadService.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/LookUpService.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/BytesChunk.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitFileFromChunks.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitFileFromLocalFile.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/Address.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/jgroups/JGroupsTransport.java trunk/diswork-fs/src/main/resources/log4j.properties trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/AbstractSplitFileTest.java trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/DistributedFileSystemTest.java trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplitFileFromChunksTest.java trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplitFileFromLocalFileTest.java Copied: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkConfig.java (from rev 19, trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/config/DisworkConfig.java) =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkConfig.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkConfig.java 2010-05-06 08:25:12 UTC (rev 23) @@ -0,0 +1,25 @@ +package org.nuiton.disworkfs; + +import java.io.File; +import java.util.Random; + +import org.nuiton.util.ApplicationConfig; + +public class DisworkConfig extends ApplicationConfig { + + public DisworkConfig() { + Random random = new Random(); + setDefaultOption("storage", "/tmp/disworkfs/storage" + random.nextInt()); + setDefaultOption("jgroups.cluster_name", "diswork-fs"); + + // replication strategy... + } + + public File getStoragePath() { + return getOptionAsFile("storage"); + } + + public String getJGroupsClusterName() { + return getOption("jgroups.cluster_name"); + } +} Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkFileSystem.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkFileSystem.java 2010-05-04 17:54:32 UTC (rev 22) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkFileSystem.java 2010-05-06 08:25:12 UTC (rev 23) @@ -8,7 +8,6 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.nuiton.disworkfs.config.DisworkConfig; import org.nuiton.disworkfs.services.DownloadService; import org.nuiton.disworkfs.services.LookUpService; import org.nuiton.disworkfs.services.UploadService; @@ -21,7 +20,7 @@ private LookUpService lookUpService; private DisworkConfig disworkConfig; - private Log log = LogFactory.getLog(DisworkFileSystem.class); + private static final Log log = LogFactory.getLog(DisworkFileSystem.class); public DisworkFileSystem(DisworkConfig disworkConfig) { @@ -57,7 +56,7 @@ log.info("trying to read " + path); - File file = new File(disworkConfig.getOption("storage"), path); + File file = new File(disworkConfig.getStoragePath(), path); log.info("trying at local file system " + file.getAbsolutePath()); if (!file.exists()) { @@ -89,7 +88,7 @@ public void write(String path, File source) throws IOException { - File target = new File(disworkConfig.getOption("storage"), path); + File target = new File(disworkConfig.getStoragePath(), path); /* if (file.exists()) { // TODO Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkServicesManager.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkServicesManager.java 2010-05-04 17:54:32 UTC (rev 22) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkServicesManager.java 2010-05-06 08:25:12 UTC (rev 23) @@ -3,9 +3,6 @@ import java.util.ArrayList; import java.util.List; -import org.jgroups.Address; -import org.jgroups.View; -import org.nuiton.disworkfs.config.DisworkConfig; import org.nuiton.disworkfs.services.AbstractDisworkService; import org.nuiton.disworkfs.transport.Message; import org.nuiton.disworkfs.transport.Receiver; Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/RunMe.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/RunMe.java 2010-05-04 17:54:32 UTC (rev 22) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/RunMe.java 2010-05-06 08:25:12 UTC (rev 23) @@ -1,6 +1,5 @@ package org.nuiton.disworkfs; -import org.nuiton.disworkfs.config.DisworkConfig; public class RunMe { Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/SimpleDownload.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/SimpleDownload.java 2010-05-04 17:54:32 UTC (rev 22) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/SimpleDownload.java 2010-05-06 08:25:12 UTC (rev 23) @@ -18,7 +18,7 @@ private LookUpService lookUpService; private String filePath; - private Log log = LogFactory.getLog(SimpleDownload.class); + private static final Log log = LogFactory.getLog(SimpleDownload.class); // TODO timeout public SimpleDownload(String filePath, LookUpService lookUpService, DownloadService downloadService) throws Exception { @@ -30,10 +30,13 @@ public boolean initiateDownload() { lookUpService.lookForFileName(filePath, this); - + // FIXME bad implementation of a timeout int numberOfSecondsWaited = 0; - while (this.fileDescription == null && numberOfSecondsWaited <= 10) { + boolean responseReceived = false; + + // wait until there is a result or timeout + while (!responseReceived && numberOfSecondsWaited <= 10) { // response not yet received, wait again... try { Thread.sleep(1000); @@ -43,27 +46,35 @@ // TODO Auto-generated catch block e.printStackTrace(); } + + responseReceived = this.fileDescription != null; + if (responseReceived) + log.info("look-up response received"); } - return fileDescription != null; + return responseReceived; } public File startDownload() { + if (log.isDebugEnabled()) + log.info("starting download for " + fileDescription.getFileName()); downloadService.startDownload(fileDescription, this); // TODO throw file not found if timeout exceed while(! downloadFinised) { try { - Thread.sleep(1000); - log.info("waiting for download complete"); + Thread.sleep(10 * 1000); + log.info("still waiting for " + fileDescription.getFileName() + " download to complete"); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } } + if (log.isDebugEnabled()) + log.info("download " + fileDescription.getFileName() + " is complete"); return new File(fileDescription.getFileName()); } Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/AbstractDisworkService.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/AbstractDisworkService.java 2010-05-04 17:54:32 UTC (rev 22) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/AbstractDisworkService.java 2010-05-06 08:25:12 UTC (rev 23) @@ -4,9 +4,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.jgroups.Address; -import org.jgroups.View; -import org.nuiton.disworkfs.config.DisworkConfig; +import org.nuiton.disworkfs.DisworkConfig; import org.nuiton.disworkfs.messages.FileRequestMessage; import org.nuiton.disworkfs.messages.FileTransferMessage; import org.nuiton.disworkfs.messages.LookUpMessage; @@ -23,7 +21,7 @@ protected DisworkConfig disworkConfig; - private Log log = LogFactory.getLog(AbstractDisworkService.class); + private static final Log log = LogFactory.getLog(AbstractDisworkService.class); public AbstractDisworkService() {} Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/DownloadService.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/DownloadService.java 2010-05-04 17:54:32 UTC (rev 22) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/DownloadService.java 2010-05-06 08:25:12 UTC (rev 23) @@ -1,7 +1,6 @@ package org.nuiton.disworkfs.services; import java.io.File; -import java.io.IOException; import java.util.HashMap; import java.util.LinkedList; import java.util.List; @@ -30,7 +29,8 @@ public void receiveFileTransferMessage(Message message) { FileTransferMessage fileTransferMessage = (FileTransferMessage) message.getContent(); - + FileDescription fileDescription = fileTransferMessage.getFileDescrition(); + if (log.isDebugEnabled()) { log.info("received file chunk " + fileTransferMessage.getFileDescrition().getFileName() @@ -39,47 +39,51 @@ + ")"); } - Long checkSum = fileTransferMessage.getFileDescrition().getFileCheckSum(); + // Before do anything, check if we already have the complete file - if (! downloadsInProgress.containsKey(checkSum)) { - // Start download + File newFile = new File(disworkConfig.getStoragePath(), fileDescription.getFileName()); + File newFileStatus = new File(disworkConfig.getStoragePath(), "." + fileDescription.getFileName() + ".index"); + + // file must exists, have the expected size and index should not exist + // TODO : use checksum ? + boolean alreadyHaveFile = newFile.exists() + && (newFile.length() == fileDescription.getTotalSize()) + && !newFileStatus.exists(); + + if (!alreadyHaveFile) { + Long checkSum = fileDescription.getFileCheckSum(); - log.info("first chunk received, initiate download"); + if (! downloadsInProgress.containsKey(checkSum)) { + // Start download - SplitFileFromChunks newSplitFile = new SplitFileFromChunks(fileTransferMessage.getFileDescrition()); + log.info("first chunk received, initiate download"); - downloadsInProgress.put(checkSum, newSplitFile); - } + SplitFileFromChunks newSplitFile = new SplitFileFromChunks(fileDescription, newFile); - // we have received a file chunk, let's add it - SplitFileFromChunks downloadingFile = downloadsInProgress.get(checkSum); + downloadsInProgress.put(checkSum, newSplitFile); + } - downloadingFile.addChunk(fileTransferMessage.getFileChunk()); + // we have received a file chunk, let's add it + SplitFileFromChunks downloadingFile = downloadsInProgress.get(checkSum); - // maybe the download is complete - if (downloadingFile.isComplete()) { - // write file to the FS - - try { - File newFile = new File(disworkConfig.getStoragePath(), fileTransferMessage.getFileDescrition().getFileName()); - downloadingFile.writeToLocalFileSystem(newFile); + downloadingFile.addChunk(fileTransferMessage.getFileChunk()); + + // maybe the download is complete + if (downloadingFile.isComplete()) { + // write file to the FS log.info("file " + newFile.getAbsolutePath() + " written"); - } catch (IOException e) { - log.error("can't write file"); - } catch (Exception e) { - log.error("trying to write data with missing chunks"); - } - finishedDownloads.add(checkSum); - downloadsInProgress.remove(checkSum); + finishedDownloads.add(checkSum); + downloadsInProgress.remove(checkSum); - this.notifyAllDownloadObserversForFile(checkSum); + this.notifyAllDownloadObserversForFile(checkSum); + } } } public void registerObserver(FileDescription fileDescription, DownloadObserver downloadObserver) { List<DownloadObserver> observersList = downloadObservers.get(fileDescription.getFileCheckSum()); - + if (observersList == null) { // it's the first observer for this download ever, let's construct a list observersList = new LinkedList<DownloadObserver>(); @@ -90,11 +94,11 @@ private void notifyAllDownloadObserversForFile(Long checksum) { List<DownloadObserver> downloadObserversForThisFile = downloadObservers.get(checksum); - + // important check : maybe no observer for this file download // so the list is null. if (downloadObserversForThisFile != null) { - + for (DownloadObserver downloadObserver : downloadObserversForThisFile) { downloadObserver.updateDownloadStatus(this); } @@ -106,27 +110,7 @@ return finished; } - public double getProgess(FileDescription fileDescription) { - // FIXME this method never return intermediate values - double progress = 0.0; - if (isFinished(fileDescription)) { - progress = 1.0; - } else { - Long checksum = fileDescription.getFileCheckSum(); - if (downloadsInProgress.containsKey(checksum)) { - // download is in progress - long actualSize = downloadsInProgress.get(checksum).getActualSize(); - long expectedSize = fileDescription.getTotalSize(); - progress = actualSize / expectedSize; - } else { - // download not started - progress = -1.0; - } - } - return progress; - } - public void startDownload(FileDescription fileDescription, DownloadObserver downloadObserver) { Message message = transport.newMulticastMessage(); message.setContent(new FileRequestMessage(fileDescription)); Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/LookUpService.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/LookUpService.java 2010-05-04 17:54:32 UTC (rev 22) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/LookUpService.java 2010-05-06 08:25:12 UTC (rev 23) @@ -23,11 +23,11 @@ private static final Log log = LogFactory.getLog(LookUpService.class); - /** - * - * @param fileName the name of the name - * @return the FileDescription or null of file have not been found - */ + /** + * + * @param fileName the name of the file to search for + * @param lookUpObserver the object to notify when LookUpResponse have been sent back + */ public void lookForFileName(String fileName, LookUpObserver lookUpObserver) { LookUpMessage lookUpMessage = new LookUpMessage(fileName); Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/BytesChunk.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/BytesChunk.java 2010-05-04 17:54:32 UTC (rev 22) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/BytesChunk.java 2010-05-06 08:25:12 UTC (rev 23) @@ -2,7 +2,6 @@ import java.io.Serializable; -// TODO really implement Serializable public class BytesChunk implements Serializable { /** Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitFileFromChunks.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitFileFromChunks.java 2010-05-04 17:54:32 UTC (rev 22) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitFileFromChunks.java 2010-05-06 08:25:12 UTC (rev 23) @@ -1,80 +1,119 @@ package org.nuiton.disworkfs.split; import java.io.File; +import java.io.FileInputStream; +import java.io.FileNotFoundException; +import java.io.FileOutputStream; import java.io.IOException; -import java.util.ArrayList; -import java.util.List; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.io.RandomAccessFile; +import java.util.BitSet; import java.util.zip.CRC32; import org.apache.commons.io.FileUtils; -import org.apache.commons.lang.NotImplementedException; public class SplitFileFromChunks { - private FileDescription fileDescription; - - private List<FileChunk> fileChunks; - + protected FileDescription fileDescription; + protected File destination; + protected File chunkStatusFile; - public SplitFileFromChunks(FileDescription fileDescription) { + public SplitFileFromChunks(FileDescription fileDescription, File destination) { this.fileDescription = fileDescription; - fileChunks = new ArrayList<FileChunk>(); + this.destination = destination; + this.chunkStatusFile = new File(destination.getParent(), "." + destination.getName() + ".index"); + + if (!chunkStatusFile.exists()) { + BitSet bitSet = new BitSet(fileDescription.getNumberOfChunks()); + bitSet.set(0, fileDescription.getNumberOfChunks()); + writeChunkStatusFile(bitSet); + bitSet = readChunkStatusFile(); + } + + if (!destination.exists()) { + try { + RandomAccessFile randomAccessFile = new RandomAccessFile(destination, "rw"); + randomAccessFile.setLength(fileDescription.getTotalSize()); + randomAccessFile.close(); + } catch (FileNotFoundException e) { + // we just checked !destination.exists() + } catch (IOException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + } } - public void addChunk(FileChunk fileChunk) { - fileChunks.add(fileChunk); - } - public boolean isComplete() { - return fileChunks.size() == fileDescription.getNumberOfChunks(); + BitSet bitSet = readChunkStatusFile(); + boolean fileIsComplete = bitSet.cardinality() == 0; + return fileIsComplete; } - public long getActualSize() { - int actualSize = 0; - for (FileChunk fileChunk : fileChunks) { - actualSize += fileChunk.getBytesChunk().getChunkSize(); + + protected BitSet readChunkStatusFile() { + BitSet bitSet = null; + try { + FileInputStream is = new FileInputStream(chunkStatusFile); + bitSet = (BitSet) new ObjectInputStream(is).readObject(); + is.close(); + } catch (IOException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } catch (ClassNotFoundException e) { + // TODO Auto-generated catch block + e.printStackTrace(); } - return actualSize; + return bitSet; } - public List<Integer> getMissingChunksIds() { - throw new NotImplementedException(); + + protected void writeChunkStatusFile(BitSet bitSet) { + try { + FileOutputStream os = new FileOutputStream(chunkStatusFile); + new ObjectOutputStream(os).writeObject(bitSet); + os.close(); + } catch (IOException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } } - public boolean writeToLocalFileSystem() throws IOException, Exception { - File newFile = new File(fileDescription.getFileName()); - return writeToLocalFileSystem(newFile); - } - - /** - * - * @return the result of the check integrity - * @throws IOException - * @throws Exception - */ - public boolean writeToLocalFileSystem(File newFile) throws IOException, Exception { + public void addChunk(FileChunk fileChunk) { + try { + RandomAccessFile randomAccessFile = new RandomAccessFile(destination, "rw"); + byte[] data = fileChunk.getBytesChunk().getData(); + int off = BytesChunk.MAX_CHUNK_SIZE * fileChunk.getBytesChunk().getChunkNumber(); + int len = fileChunk.getBytesChunk().getChunkSize(); + randomAccessFile.seek(off); + randomAccessFile.write(data, 0, len); + randomAccessFile.close(); + + // updating status + + BitSet chunkStatus = readChunkStatusFile(); + chunkStatus.clear(fileChunk.getBytesChunk().getChunkNumber()); + writeChunkStatusFile(chunkStatus); + + + if (isComplete()) { - if (fileChunks == null) { - throw new Exception("No data to write"); - } - - - long expectedSize = fileDescription.getTotalSize(); - - SplittedBytes splittedBytes = new SplittedBytes(expectedSize); - - for (FileChunk fileChunk : fileChunks) { - splittedBytes.addChunk(fileChunk.getBytesChunk()); - } - - FileUtils.writeByteArrayToFile(newFile, splittedBytes.getBytesFromChunks()); - - // checking integrity - long expectedChecksum = fileDescription.getFileCheckSum(); - long actualCheckSum = FileUtils.checksum(newFile, new CRC32()).getValue(); - - return actualCheckSum == expectedChecksum; + long expectedChecksum = fileDescription.getFileCheckSum(); + long actualCheckSum = FileUtils.checksum(destination, new CRC32()).getValue(); + + if (actualCheckSum != expectedChecksum) { + throw new IOException("checksum fail"); + } + } + } catch (FileNotFoundException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } catch (IOException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } } } Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitFileFromLocalFile.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitFileFromLocalFile.java 2010-05-04 17:54:32 UTC (rev 22) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitFileFromLocalFile.java 2010-05-06 08:25:12 UTC (rev 23) @@ -1,7 +1,9 @@ package org.nuiton.disworkfs.split; import java.io.File; +import java.io.FileNotFoundException; import java.io.IOException; +import java.io.RandomAccessFile; import java.util.ArrayList; import java.util.List; import java.util.zip.CRC32; @@ -10,53 +12,56 @@ public class SplitFileFromLocalFile { - private File localFile; + protected File source; - public SplitFileFromLocalFile(File file) { - localFile = file; + public SplitFileFromLocalFile(File source) { + this.source = source; } - + public FileDescription getFileDescription() throws IOException { - String localFileName = localFile.getPath(); - long localFileTotalSize = localFile.length(); - long localFileCheckSum = FileUtils.checksum(localFile, new CRC32()).getValue(); + String localFileName = source.getPath(); + long localFileTotalSize = source.length(); + long localFileCheckSum = FileUtils.checksum(source, new CRC32()).getValue(); return new FileDescription(localFileName, localFileTotalSize, localFileCheckSum); } - - public List<FileChunk> getAllChunks() throws IOException { - byte[] data = FileUtils.readFileToByteArray(localFile); + + public List<FileChunk> getAllChunks() throws IOException, FileNotFoundException { + + RandomAccessFile randomAccessFile = new RandomAccessFile(source, "r"); + + // preparing an empty list to store the result + List<FileChunk> result = new ArrayList<FileChunk>(); - // Creating the file chunks - SplittedBytes splittedBytes = new SplittedBytes(); - splittedBytes.setChunksFromBytes(data); + // this array will contains some bytes read from the file + byte[] read = new byte[BytesChunk.MAX_CHUNK_SIZE]; + + // chunks have to be numbered + int chunkNumber = 0; - List<FileChunk> fileChunks = new ArrayList<FileChunk>(); - for(BytesChunk bytesChunk : splittedBytes.getChunks()) { - fileChunks.add(new FileChunk(bytesChunk)); + // the last chunk will not be complete, so we have to store the + // chunkSize for each chunk + int chunkSize; + + // reading the file until the end + while ((chunkSize = randomAccessFile.read(read)) != -1) { + + // creating a FileChunk from the data read + BytesChunk bytesChunk = new BytesChunk(read); + bytesChunk.setChunkNumber(chunkNumber); + bytesChunk.setChunkSize(chunkSize); + + FileChunk fileChunk = new FileChunk(bytesChunk); + result.add(fileChunk); + + // preparing data for next iteration + read = new byte[BytesChunk.MAX_CHUNK_SIZE]; + chunkNumber += 1; } - return fileChunks; - } - - public List<FileChunk> getSomeChunks(int[] indexes) throws IOException { - // TODO implement - List<FileChunk> allChunks = this.getAllChunks(); - List<FileChunk> someChunks = new ArrayList<FileChunk>(); - for (int index : indexes) { - someChunks.add(allChunks.get(index)); - } - return someChunks; - } - - public FileChunk getChunk(int index) throws IOException { - int singletonIndex[] = { index }; + + randomAccessFile.close(); - return this.getSomeChunks(singletonIndex).get(0); - + return result; } - - public int getNumberOfChunks() throws IOException { - return this.getFileDescription().getNumberOfChunks(); - } - - + + } Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/Address.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/Address.java 2010-05-04 17:54:32 UTC (rev 22) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/Address.java 2010-05-06 08:25:12 UTC (rev 23) @@ -3,5 +3,10 @@ import java.io.Serializable; public abstract class Address implements Serializable { + + /** + * + */ + private static final long serialVersionUID = 4178387970911345672L; } Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/jgroups/JGroupsTransport.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/jgroups/JGroupsTransport.java 2010-05-04 17:54:32 UTC (rev 22) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/jgroups/JGroupsTransport.java 2010-05-06 08:25:12 UTC (rev 23) @@ -2,14 +2,15 @@ import java.io.Serializable; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.jgroups.Channel; import org.jgroups.ChannelClosedException; import org.jgroups.ChannelException; import org.jgroups.ChannelNotConnectedException; import org.jgroups.JChannel; import org.jgroups.View; -import org.nuiton.disworkfs.config.DisworkConfig; -import org.nuiton.disworkfs.services.AbstractDisworkService; +import org.nuiton.disworkfs.DisworkConfig; import org.nuiton.disworkfs.transport.Address; import org.nuiton.disworkfs.transport.Message; import org.nuiton.disworkfs.transport.Receiver; @@ -21,9 +22,11 @@ protected JChannel jChannel; protected Receiver receiver; + private static final Log log = LogFactory.getLog(JGroupsTransport.class); + /** - * This class is just a delegator. Every message received on the jChannel, - * will be sent to the Receiver who subscribed to transport + * This class is just a delegator. Every message received on this.jChannel, + * will be sent to this.receiver who subscribed to transport * @author bleny */ private class MyReceiver extends org.jgroups.ExtendedReceiverAdapter { @@ -34,16 +37,32 @@ this.tranport = tranport; } + /** + * receive a new message from the JChannel, send it to the + * transport layer + */ public void receive(org.jgroups.Message msg) { + // this method is call be the jChannel when a message is received + if (log.isDebugEnabled()) + log.info("message received from jChannel " + msg); + // let's create a new message for the transport layer Message message = tranport.newEmptyMessage(); + // setting the message attributes, according to the JGroups message message.setSource(new JGroupsAddress(msg.getSrc())); message.setSource(new JGroupsAddress(msg.getDest())); message.setContent((Serializable) msg.getObject()); + // sending our new message to the transport layer + if (log.isDebugEnabled()) + log.info("message received from jChannel sent to transport layer as " + message); receiver.receiveMessage(message); } + + public void viewAccepted(View new_view) { + log.info("now seeing " + new_view.size() + " nodes"); + } } @@ -52,49 +71,61 @@ this.disworkConfig = disworkConfig; try { jChannel = new JChannel("udp.xml"); - jChannel.connect("MonPremierGroupe"); + String clusterName = disworkConfig.getJGroupsClusterName(); + + log.info("connecting to JGroup " + clusterName); + jChannel.connect(clusterName); + // don't receive messages sent by myself jChannel.setOpt(Channel.LOCAL, new Boolean(false)); + // setting the receiver so every message received + // on jChannel will be received by the rest of + // the application MyReceiver myReceiver = new MyReceiver(this); jChannel.setReceiver(myReceiver); } catch (ChannelException e) { - // TODO Auto-generated catch block - e.printStackTrace(); + log.error("error while creating and connecting to the JGroups channel"); } } + @Override public void setReceiver(Receiver receiver) { this.receiver = receiver; } + @Override public Address getMulticastAddress() { // null is the multicast destination in JGroups return new JGroupsAddress(null); } + @Override public boolean send(Message message) { org.jgroups.Message msg = ((JGroupsMessage) message).getMessage(); + boolean success = true; try { jChannel.send(msg); } catch (ChannelNotConnectedException e) { - // TODO log - return false; + log.error("JGroups channel not connected while trying to send a message"); + success = false; } catch (ChannelClosedException e) { - // TODO log - return false; + log.error("JGroups channel was closed while trying to send a message"); + success = false; } - return true; + return success; } + @Override public Message newEmptyMessage() { Message message = new JGroupsMessage(this, new org.jgroups.Message()); message.setSource(this.getLocalAddress()); return message; } + @Override public Message newMulticastMessage() { Message message = this.newEmptyMessage(); message.setDestination(this.getMulticastAddress()); @@ -103,7 +134,6 @@ @Override public Address getLocalAddress() { - // null is the local address in JGroups return new JGroupsAddress(jChannel.getAddress()); } Modified: trunk/diswork-fs/src/main/resources/log4j.properties =================================================================== --- trunk/diswork-fs/src/main/resources/log4j.properties 2010-05-04 17:54:32 UTC (rev 22) +++ trunk/diswork-fs/src/main/resources/log4j.properties 2010-05-06 08:25:12 UTC (rev 23) @@ -6,3 +6,6 @@ log4j.appender.stdout.layout.ConversionPattern=%d %5p [%t] (%F:%L) %M - %m%n # package level log4j.logger.org.nuiton.disworkfs=DEBUG +log4j.logger.org.nuiton.disworkfs.transport=WARN +#log4j.logger.org.nuiton.disworkfs.services.DownloadService=WARN +#log4j.logger.org.nuiton.disworkfs.services.UploadService=WARN Modified: trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/AbstractSplitFileTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/AbstractSplitFileTest.java 2010-05-04 17:54:32 UTC (rev 22) +++ trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/AbstractSplitFileTest.java 2010-05-06 08:25:12 UTC (rev 23) @@ -36,21 +36,22 @@ public void setUp() throws Exception { File tempDirectory = new File(tempDirectoryPath); tempDirectory.mkdir(); - + FileUtils.forceDeleteOnExit(tempDirectory); + // creating random data for the file byte[] randomBytes = new byte[randomFileSize]; random.nextBytes(randomBytes); // dumping random data into the file File randomFile = new File(randomFilePath); - FileUtils.writeByteArrayToFile(randomFile, randomBytes); + FileUtils.writeByteArrayToFile(randomFile, randomBytes); + } @After public void tearDown() throws Exception { // cleaning - new File(randomFilePath).delete(); - new File(tempDirectoryPath).delete(); + FileUtils.forceDelete(new File(tempDirectoryPath)); } } Modified: trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/DistributedFileSystemTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/DistributedFileSystemTest.java 2010-05-04 17:54:32 UTC (rev 22) +++ trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/DistributedFileSystemTest.java 2010-05-06 08:25:12 UTC (rev 23) @@ -7,7 +7,6 @@ import org.junit.After; import org.junit.Before; import org.junit.Test; -import org.nuiton.disworkfs.config.DisworkConfig; import org.nuiton.util.FileUtil; public class DistributedFileSystemTest { @@ -27,7 +26,7 @@ /** * The file will have this fixed size */ - static protected int randomFileSize = 10 * 1000; + static protected int randomFileSize = 3000; static protected String storagePath1; static protected String storagePath2; Modified: trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplitFileFromChunksTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplitFileFromChunksTest.java 2010-05-04 17:54:32 UTC (rev 22) +++ trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplitFileFromChunksTest.java 2010-05-06 08:25:12 UTC (rev 23) @@ -1,6 +1,5 @@ package org.nuiton.disworkfs.split; -import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -15,9 +14,8 @@ public class SplitFileFromChunksTest extends AbstractSplitFileTest { - @Test - public void simpleCopy() { + public void simpleCopyRandomAccess() { try { File randomFile = new File(randomFilePath); @@ -30,20 +28,16 @@ String splitedFileCopyPath = fileDescription.getFileName() + "_copy"; fileDescription.setFileName(splitedFileCopyPath); - SplitFileFromChunks splitFileCopy = new SplitFileFromChunks(fileDescription); + SplitFileFromChunks splitFileCopy = new SplitFileFromChunks(fileDescription, new File(splitedFileCopyPath)); // here is the simple copy from fist file to second file for (FileChunk fileChunk : splitFile.getAllChunks()) { assertFalse(splitFileCopy.isComplete()); - assertTrue(splitFileCopy.getActualSize() < fileDescription.getTotalSize()); splitFileCopy.addChunk(fileChunk); } - assertEquals(splitFileCopy.getActualSize(), fileDescription.getTotalSize()); assertTrue(splitFileCopy.isComplete()); // write the copy to the FS - boolean result = splitFileCopy.writeToLocalFileSystem(); - assertTrue("checking copy integrity", result); // compare the original and the copy byte by byte try { Modified: trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplitFileFromLocalFileTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplitFileFromLocalFileTest.java 2010-05-04 17:54:32 UTC (rev 22) +++ trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplitFileFromLocalFileTest.java 2010-05-06 08:25:12 UTC (rev 23) @@ -6,6 +6,7 @@ import java.io.File; import java.io.IOException; +import java.util.List; import org.junit.Test; import org.nuiton.disworkfs.AbstractSplitFileTest; @@ -25,8 +26,16 @@ assertEquals(SplittedBytes.numberOfChunksNeededtoStore(randomFileSize), fileDescription.getNumberOfChunks()); assertEquals(fileDescription.getFileName(), randomFilePath); - splitFileFromLocalFile.getAllChunks(); + List<FileChunk> allChunks = splitFileFromLocalFile.getAllChunks(); + for (FileChunk fileChunk : allChunks) { + int chunkSize = fileChunk.getBytesChunk().getChunkSize(); + + assertTrue( + chunkSize == BytesChunk.MAX_CHUNK_SIZE + || chunkSize == randomFileSize % BytesChunk.MAX_CHUNK_SIZE); + } + } catch (IOException e) { fail(); e.printStackTrace();

1 0

r22 - trunk/diswork-fs/src/main/java/org/nuiton/disworkfs
by bleny＠users.nuiton.org May 4, 2010

May 4, 2010

Author: bleny Date: 2010-05-04 19:54:32 +0200 (Tue, 04 May 2010) New Revision: 22 Url: http://nuiton.org/repositories/revision/diswork/22 Log: timeout pour que hudson broute pas trop Removed: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/JGroupsTest.java Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkFileSystem.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/SimpleDownload.java Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkFileSystem.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkFileSystem.java 2010-05-04 17:33:12 UTC (rev 21) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkFileSystem.java 2010-05-04 17:54:32 UTC (rev 22) @@ -66,8 +66,9 @@ try { SimpleDownload simpleDownload = new SimpleDownload(path, lookUpService, downloadService); - simpleDownload.initiateDownload(); - simpleDownload.startDownload(); + boolean fileFound = simpleDownload.initiateDownload(); + if (fileFound) + simpleDownload.startDownload(); } catch (Exception e) { // TODO Auto-generated catch block e.printStackTrace(); Deleted: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/JGroupsTest.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/JGroupsTest.java 2010-05-04 17:33:12 UTC (rev 21) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/JGroupsTest.java 2010-05-04 17:54:32 UTC (rev 22) @@ -1,35 +0,0 @@ -package org.nuiton.disworkfs; - -import org.jgroups.Channel; -import org.jgroups.ChannelException; -import org.jgroups.JChannel; -import org.jgroups.Message; -import org.jgroups.ReceiverAdapter; - -public class JGroupsTest extends ReceiverAdapter { - - public JGroupsTest() { - try { - JChannel jChannel = new JChannel("udp.xml"); - jChannel.connect("MonPremierGroupe"); - System.out.println("Channel.LOCAL was " + jChannel.getOpt(Channel.LOCAL)); - jChannel.setOpt(Channel.LOCAL, new Boolean(false)); - System.out.println("Channel.LOCAL is now " + jChannel.getOpt(Channel.LOCAL)); - jChannel.setReceiver(this); - jChannel.send(null, null, new String("Hello !")); - } catch (ChannelException e) { - e.printStackTrace(); - } - } - - public static void main(String args[]) { - new JGroupsTest(); - } - - @Override - public void receive(Message msg) { - System.out.println(msg.getObject()); - - } - -} Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/SimpleDownload.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/SimpleDownload.java 2010-05-04 17:33:12 UTC (rev 21) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/SimpleDownload.java 2010-05-04 17:54:32 UTC (rev 22) @@ -27,21 +27,25 @@ this.lookUpService = lookUpService; } - public void initiateDownload() { + public boolean initiateDownload() { lookUpService.lookForFileName(filePath, this); - while (this.fileDescription == null) { + int numberOfSecondsWaited = 0; + + while (this.fileDescription == null && numberOfSecondsWaited <= 10) { // response not yet received, wait again... try { Thread.sleep(1000); log.info("waiting for look-up response"); + numberOfSecondsWaited += 1; } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } } + return fileDescription != null; }

1 0

r21 - in trunk/diswork-fs/src: main/java/org/nuiton/disworkfs main/java/org/nuiton/disworkfs/services main/java/org/nuiton/disworkfs/split test/java/org/nuiton/disworkfs
by bleny＠users.nuiton.org May 4, 2010

May 4, 2010

Author: bleny Date: 2010-05-04 19:33:12 +0200 (Tue, 04 May 2010) New Revision: 21 Url: http://nuiton.org/repositories/revision/diswork/21 Log: -sysout +log Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkFileSystem.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkServicesManager.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/SimpleDownload.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/AbstractDisworkService.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/DownloadService.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/LookUpService.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitFileFromLocalFile.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplittedBytes.java trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/DistributedFileSystemTest.java Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkFileSystem.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkFileSystem.java 2010-05-04 17:02:50 UTC (rev 20) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkFileSystem.java 2010-05-04 17:33:12 UTC (rev 21) @@ -5,9 +5,9 @@ import java.io.FileOutputStream; import java.io.IOException; import java.io.OutputStream; -import java.net.InetAddress; -import java.net.UnknownHostException; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.nuiton.disworkfs.config.DisworkConfig; import org.nuiton.disworkfs.services.DownloadService; import org.nuiton.disworkfs.services.LookUpService; @@ -21,20 +21,23 @@ private LookUpService lookUpService; private DisworkConfig disworkConfig; + private Log log = LogFactory.getLog(DisworkFileSystem.class); + public DisworkFileSystem(DisworkConfig disworkConfig) { this.disworkConfig = disworkConfig; // System.getProperty("user.home") + /* String localIp; try { localIp = InetAddress.getLocalHost().getHostAddress(); - System.out.println(localIp); + log.info("local IP is " + localIp); } catch (UnknownHostException e) { - // TODO Auto-generated catch block - e.printStackTrace(); + log.error("can't get local IP"); } + */ // System.setProperty("jgroups.bind_addr", localIp); // System.setProperty("jgroups.tcpping.initial_hosts", localIp); // System.setProperty("jgroups.udp.mcast_addr", "224.0.0.150"); @@ -52,17 +55,16 @@ public OutputStream read(String path) { - System.out.println("trying to read " + path); + log.info("trying to read " + path); File file = new File(disworkConfig.getOption("storage"), path); - System.out.println("trying at local file system " + file.getAbsolutePath()); + log.info("trying at local file system " + file.getAbsolutePath()); if (!file.exists()) { // the file is not available // let's download it try { - SimpleDownload simpleDownload = new SimpleDownload(path, lookUpService, downloadService); simpleDownload.initiateDownload(); simpleDownload.startDownload(); @@ -85,11 +87,8 @@ } public void write(String path, File source) throws IOException { - System.out.println(disworkConfig.getOption("storage")); - System.out.println(path); File target = new File(disworkConfig.getOption("storage"), path); - System.out.println("writing to local file " + target.getAbsolutePath()); /* if (file.exists()) { // TODO Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkServicesManager.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkServicesManager.java 2010-05-04 17:02:50 UTC (rev 20) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkServicesManager.java 2010-05-04 17:33:12 UTC (rev 21) @@ -19,8 +19,6 @@ */ private List<AbstractDisworkService> services = new ArrayList<AbstractDisworkService>(); - // private JChannel jChannel; - protected Transport transport; private DisworkConfig disworkConfig; @@ -29,31 +27,11 @@ this.disworkConfig = disworkConfig; transport = new JGroupsTransport(disworkConfig); - /* - try { - jChannel = new JChannel("udp.xml"); - jChannel.connect("MonPremierGroupe"); - jChannel.setReceiver(this); - - // don't receive messages sent by myself - jChannel.setOpt(Channel.LOCAL, new Boolean(false)); - } catch (ChannelException e) { - // TODO Auto-generated catch block - e.printStackTrace(); - }*/ + transport.setReceiver(this); - System.out.println("local address = " + transport.getLocalAddress()); } - /* - public void receive(Message msg) { - // TODO log - for (AbstractDisworkService service : services) { - service.receive(msg); - } - } -*/ public void register(AbstractDisworkService service) { // dependency injection, the service need a channel to send a message @@ -73,18 +51,6 @@ public void unRegister(AbstractDisworkService service) { services.remove(service); } - - public void viewAccepted(View new_view) { - for (AbstractDisworkService service : services) { - service.viewAccepted(new_view); - } - } - - public void suspect(Address suspected_mbr) { - for (AbstractDisworkService service : services) { - service.suspect(suspected_mbr); - } - } @Override @@ -93,10 +59,4 @@ service.receiveMessage(message); } } - - /* - public JChannel getJChannel() { - return this.jChannel; - } - */ } \ No newline at end of file Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/SimpleDownload.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/SimpleDownload.java 2010-05-04 17:02:50 UTC (rev 20) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/SimpleDownload.java 2010-05-04 17:33:12 UTC (rev 21) @@ -2,6 +2,8 @@ import java.io.File; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.nuiton.disworkfs.services.DownloadService; import org.nuiton.disworkfs.services.LookUpService; import org.nuiton.disworkfs.split.FileDescription; @@ -16,6 +18,8 @@ private LookUpService lookUpService; private String filePath; + private Log log = LogFactory.getLog(SimpleDownload.class); + // TODO timeout public SimpleDownload(String filePath, LookUpService lookUpService, DownloadService downloadService) throws Exception { this.filePath = filePath; @@ -31,7 +35,7 @@ // response not yet received, wait again... try { Thread.sleep(1000); - System.out.println("waiting for response"); + log.info("waiting for look-up response"); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); @@ -48,8 +52,8 @@ // TODO throw file not found if timeout exceed while(! downloadFinised) { try { - Thread.sleep(500); - System.out.println("waiting for download"); + Thread.sleep(1000); + log.info("waiting for download complete"); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); @@ -63,7 +67,6 @@ @Override public void updateDownloadStatus(DownloadService downloadService) { synchronized (downloadFinised) { - System.out.println("progess is now " + downloadService.getProgess(fileDescription)); downloadFinised = downloadService.isFinished(fileDescription); } } @@ -71,7 +74,6 @@ @Override public void receiveResult(FileDescription fileDescription) { this.fileDescription = fileDescription; - System.out.println("result received"); } } Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/AbstractDisworkService.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/AbstractDisworkService.java 2010-05-04 17:02:50 UTC (rev 20) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/AbstractDisworkService.java 2010-05-04 17:33:12 UTC (rev 21) @@ -2,6 +2,8 @@ import java.io.Serializable; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.jgroups.Address; import org.jgroups.View; import org.nuiton.disworkfs.config.DisworkConfig; @@ -20,37 +22,15 @@ protected Transport transport; protected DisworkConfig disworkConfig; + + private Log log = LogFactory.getLog(AbstractDisworkService.class); public AbstractDisworkService() {} - - /* - public AbstractDisworkService(JChannel jChannel) { - this.jChannel = jChannel; - } - */ public AbstractDisworkService(Transport transport) { this.transport = transport; } - /* - public void receive(Message msg) { - Object obj = msg.getObject(); - if (obj instanceof LookUpMessage) { - receiveLookUpMessage(msg); - } else if (obj instanceof LookUpResponseMessage) { - receiveLookUpResponseMessage(msg); - } else if (obj instanceof FileRequestMessage) { - receiveFileRequestMessage(msg); - } else if (obj instanceof FileTransferMessage) { - receiveFileTransferMessage(msg); - } else { - System.out.println("unknow message received"); - System.out.println(obj); - } - } - */ - @Override public void receiveMessage(Message message) { @@ -64,8 +44,7 @@ } else if (obj instanceof FileTransferMessage) { receiveFileTransferMessage(message); } else { - System.out.println("unknow message received"); - System.out.println(obj); + log.error("unknow message received"); } } @@ -73,30 +52,17 @@ public void receiveLookUpResponseMessage(Message msg) {} public void receiveFileRequestMessage(Message msg) {} public void receiveFileTransferMessage(Message msg) {} - - public void viewAccepted(View new_view) { - } - - public void suspect(Address suspected_mbr) { - } - - /* - public void setJChannel(JChannel jChannel) { - this.jChannel = jChannel; - } - */ - + public void setTransport(Transport transport) { this.transport = transport; } - @Override - public void run() { - } - public void setDisworkConfig(DisworkConfig disworkConfig) { this.disworkConfig = disworkConfig; } + + @Override + public void run() {} } Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/DownloadService.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/DownloadService.java 2010-05-04 17:02:50 UTC (rev 20) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/DownloadService.java 2010-05-04 17:33:12 UTC (rev 21) @@ -63,7 +63,7 @@ try { File newFile = new File(disworkConfig.getStoragePath(), fileTransferMessage.getFileDescrition().getFileName()); downloadingFile.writeToLocalFileSystem(newFile); - System.out.println("file written"); + log.info("file " + newFile.getAbsolutePath() + " written"); } catch (IOException e) { log.error("can't write file"); } catch (Exception e) { Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/LookUpService.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/LookUpService.java 2010-05-04 17:02:50 UTC (rev 20) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/LookUpService.java 2010-05-04 17:33:12 UTC (rev 21) @@ -29,21 +29,15 @@ * @return the FileDescription or null of file have not been found */ public void lookForFileName(String fileName, LookUpObserver lookUpObserver) { - // TODO use a MessageDispatcher - // System.out.println("lookForFileName(" + fileName + ")"); LookUpMessage lookUpMessage = new LookUpMessage(fileName); - /* - Message message = new Message(null, null, lookUpMessage); - */ Message message = transport.newMulticastMessage(); message.setContent(lookUpMessage); message.send(); log.info("look-up message sent for " + fileName); requestToRequester.put(fileName, lookUpObserver); - // System.out.println("expected == " + fileName); } @Override @@ -53,7 +47,6 @@ String requestedFileName = lookUpResponse.getFileDescription().getFileName(); - // System.out.println("actual == " + requestedFileName); if (requestToRequester.containsKey(requestedFileName)) { LookUpObserver requester = requestToRequester.get(requestedFileName); requester.receiveResult(lookUpResponse.getFileDescription()); Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitFileFromLocalFile.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitFileFromLocalFile.java 2010-05-04 17:02:50 UTC (rev 20) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitFileFromLocalFile.java 2010-05-04 17:33:12 UTC (rev 21) @@ -17,7 +17,6 @@ } public FileDescription getFileDescription() throws IOException { - System.out.println("returning a file description with " + localFile.getPath()); String localFileName = localFile.getPath(); long localFileTotalSize = localFile.length(); long localFileCheckSum = FileUtils.checksum(localFile, new CRC32()).getValue(); Modified: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplittedBytes.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplittedBytes.java 2010-05-04 17:02:50 UTC (rev 20) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplittedBytes.java 2010-05-04 17:33:12 UTC (rev 21) @@ -84,22 +84,18 @@ if (data.length > numberOfChunks * BytesChunk.MAX_CHUNK_SIZE) { numberOfChunks += 1; } - - // System.out.println("numberOfChunks = " + numberOfChunks); - + for (int currentChunk = 0; currentChunk < numberOfChunks; ++currentChunk) { - - // System.out.println("currentChunk = " + currentChunk); - + int numberOfLastBytes = BytesChunk.MAX_CHUNK_SIZE; if (data.length - (currentChunk * BytesChunk.MAX_CHUNK_SIZE) < BytesChunk.MAX_CHUNK_SIZE) { - // System.out.println("last chunk"); + // last chunk numberOfLastBytes = data.length - currentChunk * BytesChunk.MAX_CHUNK_SIZE; } - // System.out.println("reading " + numberOfLastBytes + " bytes"); + // reading numberOfLastBytes bytes byte[] chunkData = new byte[BytesChunk.MAX_CHUNK_SIZE]; @@ -112,11 +108,6 @@ this.chunks.add(bytesChunk); totalSize += numberOfLastBytes; - /* - for (int i = 0; i < chunkData.length; i++) { - System.out.println("chunkData[" + i + "] = " + chunkData[i]); - } - */ } } @@ -135,22 +126,9 @@ int bytesReads = 0; totalSize = 0; for (BytesChunk fileChunk : this.getChunks()) { - // System.out.println("reading chunks n° " + - // fileChunk.getChunkNumber()); - // System.out.println("reading " + fileChunk.getChunkSize()); - - // System.out.println("writing at "+fileChunk.getChunkSize()+" bytes to data["+bytesReads+"]" - // + fileChunk.getData()); System.arraycopy(fileChunk.getData(), 0, data, bytesReads, fileChunk.getChunkSize()); - /* - for (int i = 0; i < fileChunk.getChunkSize(); i++) { - byte[] tempdata = fileChunk.getData(); - System.out.println("tempdata[" + i + "] = " + tempdata[i]); - } - */ - bytesReads += fileChunk.getChunkSize(); totalSize += bytesReads; } @@ -163,7 +141,6 @@ } public boolean isComplete() { - // return this.missingChunks().isEmpty(); return totalSize == this.getActualSize(); } Modified: trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/DistributedFileSystemTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/DistributedFileSystemTest.java 2010-05-04 17:02:50 UTC (rev 20) +++ trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/DistributedFileSystemTest.java 2010-05-04 17:33:12 UTC (rev 21) @@ -27,7 +27,7 @@ /** * The file will have this fixed size */ - static protected int randomFileSize = 10 * 1000 * 1000; + static protected int randomFileSize = 10 * 1000; static protected String storagePath1; static protected String storagePath2;

1 0

r20 - trunk/diswork-fs
by bleny＠users.nuiton.org May 4, 2010

May 4, 2010

Author: bleny Date: 2010-05-04 19:02:50 +0200 (Tue, 04 May 2010) New Revision: 20 Url: http://nuiton.org/repositories/revision/diswork/20 Log: menage fichiers eclipse Removed: trunk/diswork-fs/.settings/

1 0

May 4, 2010

Author: bleny Date: 2010-05-04 18:58:34 +0200 (Tue, 04 May 2010) New Revision: 19 Url: http://nuiton.org/repositories/revision/diswork/19 Log: diswork fs, oublie au dernier commit :-( Added: trunk/diswork-fs/ trunk/diswork-fs/.settings/ trunk/diswork-fs/.settings/org.maven.ide.eclipse.prefs trunk/diswork-fs/LICENSE.txt trunk/diswork-fs/README.txt trunk/diswork-fs/changelog.txt trunk/diswork-fs/pom.xml trunk/diswork-fs/src/ trunk/diswork-fs/src/main/ trunk/diswork-fs/src/main/java/ trunk/diswork-fs/src/main/java/org/ trunk/diswork-fs/src/main/java/org/nuiton/ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DistributedFileSystem.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkFileSystem.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkServicesManager.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DownloadObserver.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/JGroupsTest.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/LookUpObserver.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/RunMe.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/SimpleDownload.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/config/ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/config/DisworkConfig.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/messages/ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/messages/FileRequestMessage.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/messages/FileTransferMessage.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/messages/LookUpMessage.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/messages/LookUpResponseMessage.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/AbstractDisworkService.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/DownloadService.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/LookUpService.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/UploadService.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/BytesChunk.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/FileChunk.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/FileDescription.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitFileFromChunks.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitFileFromLocalFile.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitedFile.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplittedBytes.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/Address.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/Message.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/Receiver.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/Transport.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/jgroups/ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/jgroups/JGroupsAddress.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/jgroups/JGroupsMessage.java trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/jgroups/JGroupsTransport.java trunk/diswork-fs/src/main/resources/ trunk/diswork-fs/src/main/resources/log4j.properties trunk/diswork-fs/src/test/ trunk/diswork-fs/src/test/java/ trunk/diswork-fs/src/test/java/org/ trunk/diswork-fs/src/test/java/org/nuiton/ trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/ trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/AbstractSplitFileTest.java trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/AllwaysReplyToLookUpTest.java trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/DistributedFileSystemTest.java trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/ trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplitFileFromChunksTest.java trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplitFileFromLocalFileTest.java trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplitedFileTest.java trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplittedBytesTest.java Added: trunk/diswork-fs/.settings/org.maven.ide.eclipse.prefs =================================================================== --- trunk/diswork-fs/.settings/org.maven.ide.eclipse.prefs (rev 0) +++ trunk/diswork-fs/.settings/org.maven.ide.eclipse.prefs 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,9 @@ +#Wed Apr 28 15:20:43 CEST 2010 +activeProfiles= +eclipse.preferences.version=1 +fullBuildGoals=process-test-resources +includeModules=false +resolveWorkspaceProjects=true +resourceFilterGoals=process-resources resources\:testResources +skipCompilerPlugin=true +version=1 Added: trunk/diswork-fs/LICENSE.txt =================================================================== --- trunk/diswork-fs/LICENSE.txt (rev 0) +++ trunk/diswork-fs/LICENSE.txt 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,166 @@ + GNU LESSER GENERAL PUBLIC LICENSE + Version 3, 29 June 2007 + + Copyright (C) 2007 Free Software Foundation, Inc. <http://fsf.org/> + Everyone is permitted to copy and distribute verbatim copies + of this license document, but changing it is not allowed. + + + This version of the GNU Lesser General Public License incorporates +the terms and conditions of version 3 of the GNU General Public +License, supplemented by the additional permissions listed below. + + 0. Additional Definitions. + + As used herein, "this License" refers to version 3 of the GNU Lesser +General Public License, and the "GNU GPL" refers to version 3 of the GNU +General Public License. + + "The Library" refers to a covered work governed by this License, +other than an Application or a Combined Work as defined below. + + An "Application" is any work that makes use of an interface provided +by the Library, but which is not otherwise based on the Library. +Defining a subclass of a class defined by the Library is deemed a mode +of using an interface provided by the Library. + + A "Combined Work" is a work produced by combining or linking an +Application with the Library. The particular version of the Library +with which the Combined Work was made is also called the "Linked +Version". + + The "Minimal Corresponding Source" for a Combined Work means the +Corresponding Source for the Combined Work, excluding any source code +for portions of the Combined Work that, considered in isolation, are +based on the Application, and not on the Linked Version. + + The "Corresponding Application Code" for a Combined Work means the +object code and/or source code for the Application, including any data +and utility programs needed for reproducing the Combined Work from the +Application, but excluding the System Libraries of the Combined Work. + + 1. Exception to Section 3 of the GNU GPL. + + You may convey a covered work under sections 3 and 4 of this License +without being bound by section 3 of the GNU GPL. + + 2. Conveying Modified Versions. + + If you modify a copy of the Library, and, in your modifications, a +facility refers to a function or data to be supplied by an Application +that uses the facility (other than as an argument passed when the +facility is invoked), then you may convey a copy of the modified +version: + + a) under this License, provided that you make a good faith effort to + ensure that, in the event an Application does not supply the + function or data, the facility still operates, and performs + whatever part of its purpose remains meaningful, or + + b) under the GNU GPL, with none of the additional permissions of + this License applicable to that copy. + + 3. Object Code Incorporating Material from Library Header Files. + + The object code form of an Application may incorporate material from +a header file that is part of the Library. You may convey such object +code under terms of your choice, provided that, if the incorporated +material is not limited to numerical parameters, data structure +layouts and accessors, or small macros, inline functions and templates +(ten or fewer lines in length), you do both of the following: + + a) Give prominent notice with each copy of the object code that the + Library is used in it and that the Library and its use are + covered by this License. + + b) Accompany the object code with a copy of the GNU GPL and this license + document. + + 4. Combined Works. + + You may convey a Combined Work under terms of your choice that, +taken together, effectively do not restrict modification of the +portions of the Library contained in the Combined Work and reverse +engineering for debugging such modifications, if you also do each of +the following: + + a) Give prominent notice with each copy of the Combined Work that + the Library is used in it and that the Library and its use are + covered by this License. + + b) Accompany the Combined Work with a copy of the GNU GPL and this license + document. + + c) For a Combined Work that displays copyright notices during + execution, include the copyright notice for the Library among + these notices, as well as a reference directing the user to the + copies of the GNU GPL and this license document. + + d) Do one of the following: + + 0) Convey the Minimal Corresponding Source under the terms of this + License, and the Corresponding Application Code in a form + suitable for, and under terms that permit, the user to + recombine or relink the Application with a modified version of + the Linked Version to produce a modified Combined Work, in the + manner specified by section 6 of the GNU GPL for conveying + Corresponding Source. + + 1) Use a suitable shared library mechanism for linking with the + Library. A suitable mechanism is one that (a) uses at run time + a copy of the Library already present on the user's computer + system, and (b) will operate properly with a modified version + of the Library that is interface-compatible with the Linked + Version. + + e) Provide Installation Information, but only if you would otherwise + be required to provide such information under section 6 of the + GNU GPL, and only to the extent that such information is + necessary to install and execute a modified version of the + Combined Work produced by recombining or relinking the + Application with a modified version of the Linked Version. (If + you use option 4d0, the Installation Information must accompany + the Minimal Corresponding Source and Corresponding Application + Code. If you use option 4d1, you must provide the Installation + Information in the manner specified by section 6 of the GNU GPL + for conveying Corresponding Source.) + + 5. Combined Libraries. + + You may place library facilities that are a work based on the +Library side by side in a single library together with other library +facilities that are not Applications and are not covered by this +License, and convey such a combined library under terms of your +choice, if you do both of the following: + + a) Accompany the combined library with a copy of the same work based + on the Library, uncombined with any other library facilities, + conveyed under the terms of this License. + + b) Give prominent notice with the combined library that part of it + is a work based on the Library, and explaining where to find the + accompanying uncombined form of the same work. + + 6. Revised Versions of the GNU Lesser General Public License. + + The Free Software Foundation may publish revised and/or new versions +of the GNU Lesser General Public License from time to time. Such new +versions will be similar in spirit to the present version, but may +differ in detail to address new problems or concerns. + + Each version is given a distinguishing version number. If the +Library as you received it specifies that a certain numbered version +of the GNU Lesser General Public License "or any later version" +applies to it, you have the option of following the terms and +conditions either of that published version or of any later version +published by the Free Software Foundation. If the Library as you +received it does not specify a version number of the GNU Lesser +General Public License, you may choose any version of the GNU Lesser +General Public License ever published by the Free Software Foundation. + + If the Library as you received it specifies that a proxy can decide +whether future versions of the GNU Lesser General Public License shall +apply, that proxy's public statement of acceptance of any version is +permanent authorization for you to choose that version for the +Library. + Added: trunk/diswork-fs/pom.xml =================================================================== --- trunk/diswork-fs/pom.xml (rev 0) +++ trunk/diswork-fs/pom.xml 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,87 @@ +<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd"> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.nuiton</groupId> + <artifactId>diswork</artifactId> + <version>0.0.1-SNAPSHOT</version> + </parent> + + <groupId>org.nuiton.diswork</groupId> + <artifactId>diswork-fs</artifactId> + + <packaging>jar</packaging> + <version>0.0.1-SNAPSHOT</version> + <name>disworkfs</name> + <url>http://maven.apache.org</url> + <dependencies> + <dependency> + <groupId>commons-logging</groupId> + <artifactId>commons-logging</artifactId> + <version>1.1.1</version> + <scope>compile</scope> + </dependency> + <dependency> + <groupId>log4j</groupId> + <artifactId>log4j</artifactId> + <version>1.2.14</version> + <scope>test</scope> + </dependency> + <dependency> + <groupId>jgroups</groupId> + <artifactId>jgroups</artifactId> + <version>2.9.0.GA</version> + <scope>compile</scope> + </dependency> + <dependency> + <groupId>commons-io</groupId> + <artifactId>commons-io</artifactId> + <version>1.4</version> + <type>jar</type> + <scope>compile</scope> + </dependency> + <dependency> + <groupId>commons-lang</groupId> + <artifactId>commons-lang</artifactId> + <version>2.5</version> + <type>jar</type> + <scope>compile</scope> + </dependency> + <dependency> + <groupId>org.nuiton</groupId> + <artifactId>nuiton-utils</artifactId> + <version>1.2.2</version> + <type>jar</type> + <scope>compile</scope> + </dependency> + +  + <dependency> + <groupId>junit</groupId> + <artifactId>junit</artifactId> + <version>4.8.1</version> + <type>jar</type> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.mockito</groupId> + <artifactId>mockito-all</artifactId> + <version>1.8.4</version> + <type>jar</type> + <scope>test</scope> + </dependency> + </dependencies> + <build> + <plugins> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-compiler-plugin</artifactId> + <version>2.0.2</version> + <configuration> + <source>1.6</source> + <target>1.6</target> + </configuration> + </plugin> + </plugins> + </build> +</project> Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DistributedFileSystem.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DistributedFileSystem.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DistributedFileSystem.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,15 @@ +package org.nuiton.disworkfs; + +import java.io.File; + +public interface DistributedFileSystem { + + public boolean exists(String path); + + public void write(String path, File source); + + public File read(String path); + + public void remove(String path); + +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkFileSystem.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkFileSystem.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkFileSystem.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,102 @@ +package org.nuiton.disworkfs; + +import java.io.File; +import java.io.FileNotFoundException; +import java.io.FileOutputStream; +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.UnknownHostException; + +import org.nuiton.disworkfs.config.DisworkConfig; +import org.nuiton.disworkfs.services.DownloadService; +import org.nuiton.disworkfs.services.LookUpService; +import org.nuiton.disworkfs.services.UploadService; +import org.nuiton.util.FileUtil; + +public class DisworkFileSystem { + + private DownloadService downloadService; + private UploadService uploadService; + private LookUpService lookUpService; + private DisworkConfig disworkConfig; + + public DisworkFileSystem(DisworkConfig disworkConfig) { + + this.disworkConfig = disworkConfig; + + // System.getProperty("user.home") + + String localIp; + try { + localIp = InetAddress.getLocalHost().getHostAddress(); + System.out.println(localIp); + } catch (UnknownHostException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + // System.setProperty("jgroups.bind_addr", localIp); + // System.setProperty("jgroups.tcpping.initial_hosts", localIp); + // System.setProperty("jgroups.udp.mcast_addr", "224.0.0.150"); + // System.setProperty("java.net.preferIPv4Stack", "true"); + + DisworkServicesManager disworkServicesManager = new DisworkServicesManager(disworkConfig); + uploadService = new UploadService(); + disworkServicesManager.register(uploadService); + downloadService = new DownloadService(); + disworkServicesManager.register(downloadService); + lookUpService = new LookUpService(); + disworkServicesManager.register(lookUpService); + + } + + public OutputStream read(String path) { + + System.out.println("trying to read " + path); + + File file = new File(disworkConfig.getOption("storage"), path); + System.out.println("trying at local file system " + file.getAbsolutePath()); + + if (!file.exists()) { + // the file is not available + // let's download it + + try { + + SimpleDownload simpleDownload = new SimpleDownload(path, lookUpService, downloadService); + simpleDownload.initiateDownload(); + simpleDownload.startDownload(); + } catch (Exception e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + + } + + OutputStream os = null; + try { + os = new FileOutputStream(file); + } catch (FileNotFoundException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + + return os; + } + + public void write(String path, File source) throws IOException { + System.out.println(disworkConfig.getOption("storage")); + System.out.println(path); + + File target = new File(disworkConfig.getOption("storage"), path); + System.out.println("writing to local file " + target.getAbsolutePath()); + /* + if (file.exists()) { + // TODO + throw new Exception("fichier existe deja"); + } + */ + + FileUtil.copy(source, target); + } +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkServicesManager.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkServicesManager.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkServicesManager.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,102 @@ +package org.nuiton.disworkfs; + +import java.util.ArrayList; +import java.util.List; + +import org.jgroups.Address; +import org.jgroups.View; +import org.nuiton.disworkfs.config.DisworkConfig; +import org.nuiton.disworkfs.services.AbstractDisworkService; +import org.nuiton.disworkfs.transport.Message; +import org.nuiton.disworkfs.transport.Receiver; +import org.nuiton.disworkfs.transport.Transport; +import org.nuiton.disworkfs.transport.jgroups.JGroupsTransport; + +public class DisworkServicesManager implements Receiver { + + /** + * all the running services + */ + private List<AbstractDisworkService> services = new ArrayList<AbstractDisworkService>(); + + // private JChannel jChannel; + + protected Transport transport; + + private DisworkConfig disworkConfig; + + public DisworkServicesManager(DisworkConfig disworkConfig) { + this.disworkConfig = disworkConfig; + + transport = new JGroupsTransport(disworkConfig); + /* + try { + jChannel = new JChannel("udp.xml"); + jChannel.connect("MonPremierGroupe"); + jChannel.setReceiver(this); + + // don't receive messages sent by myself + jChannel.setOpt(Channel.LOCAL, new Boolean(false)); + } catch (ChannelException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + }*/ + transport.setReceiver(this); + + System.out.println("local address = " + transport.getLocalAddress()); + } + + /* + public void receive(Message msg) { + // TODO log + for (AbstractDisworkService service : services) { + service.receive(msg); + } + } +*/ + public void register(AbstractDisworkService service) { + + // dependency injection, the service need a channel to send a message + // service.setJChannel(jChannel); + service.setTransport(transport); + // ... and the disworkConfig + service.setDisworkConfig(disworkConfig); + + // each service is run in his own thread + Thread thread = new Thread(service); + thread.start(); + + // add this service to the list of running services + services.add(service); + } + + public void unRegister(AbstractDisworkService service) { + services.remove(service); + } + + public void viewAccepted(View new_view) { + for (AbstractDisworkService service : services) { + service.viewAccepted(new_view); + } + } + + public void suspect(Address suspected_mbr) { + for (AbstractDisworkService service : services) { + service.suspect(suspected_mbr); + } + } + + + @Override + public void receiveMessage(Message message) { + for (AbstractDisworkService service : services) { + service.receiveMessage(message); + } + } + + /* + public JChannel getJChannel() { + return this.jChannel; + } + */ +} \ No newline at end of file Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DownloadObserver.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DownloadObserver.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DownloadObserver.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,9 @@ +package org.nuiton.disworkfs; + +import org.nuiton.disworkfs.services.DownloadService; + +public interface DownloadObserver { + + public void updateDownloadStatus(DownloadService downloadService); + +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/JGroupsTest.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/JGroupsTest.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/JGroupsTest.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,35 @@ +package org.nuiton.disworkfs; + +import org.jgroups.Channel; +import org.jgroups.ChannelException; +import org.jgroups.JChannel; +import org.jgroups.Message; +import org.jgroups.ReceiverAdapter; + +public class JGroupsTest extends ReceiverAdapter { + + public JGroupsTest() { + try { + JChannel jChannel = new JChannel("udp.xml"); + jChannel.connect("MonPremierGroupe"); + System.out.println("Channel.LOCAL was " + jChannel.getOpt(Channel.LOCAL)); + jChannel.setOpt(Channel.LOCAL, new Boolean(false)); + System.out.println("Channel.LOCAL is now " + jChannel.getOpt(Channel.LOCAL)); + jChannel.setReceiver(this); + jChannel.send(null, null, new String("Hello !")); + } catch (ChannelException e) { + e.printStackTrace(); + } + } + + public static void main(String args[]) { + new JGroupsTest(); + } + + @Override + public void receive(Message msg) { + System.out.println(msg.getObject()); + + } + +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/LookUpObserver.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/LookUpObserver.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/LookUpObserver.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,9 @@ +package org.nuiton.disworkfs; + +import org.nuiton.disworkfs.split.FileDescription; + +public interface LookUpObserver { + + public void receiveResult(FileDescription fileDescription); + +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/RunMe.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/RunMe.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/RunMe.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,13 @@ +package org.nuiton.disworkfs; + +import org.nuiton.disworkfs.config.DisworkConfig; + +public class RunMe { + + public static void main(String args[]) { + DisworkConfig disworkConfig = new DisworkConfig(); + + DisworkFileSystem disworkFileSystem = new DisworkFileSystem(disworkConfig); + } + +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/SimpleDownload.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/SimpleDownload.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/SimpleDownload.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,77 @@ +package org.nuiton.disworkfs; + +import java.io.File; + +import org.nuiton.disworkfs.services.DownloadService; +import org.nuiton.disworkfs.services.LookUpService; +import org.nuiton.disworkfs.split.FileDescription; + + + +public class SimpleDownload implements DownloadObserver, LookUpObserver { + + private Boolean downloadFinised = false; + private FileDescription fileDescription = null; + private DownloadService downloadService; + private LookUpService lookUpService; + private String filePath; + + // TODO timeout + public SimpleDownload(String filePath, LookUpService lookUpService, DownloadService downloadService) throws Exception { + this.filePath = filePath; + this.downloadService = downloadService; + this.lookUpService = lookUpService; + } + + public void initiateDownload() { + + lookUpService.lookForFileName(filePath, this); + + while (this.fileDescription == null) { + // response not yet received, wait again... + try { + Thread.sleep(1000); + System.out.println("waiting for response"); + } catch (InterruptedException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + } + + } + + + public File startDownload() { + + downloadService.startDownload(fileDescription, this); + + // TODO throw file not found if timeout exceed + while(! downloadFinised) { + try { + Thread.sleep(500); + System.out.println("waiting for download"); + } catch (InterruptedException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + } + + return new File(fileDescription.getFileName()); + } + + + @Override + public void updateDownloadStatus(DownloadService downloadService) { + synchronized (downloadFinised) { + System.out.println("progess is now " + downloadService.getProgess(fileDescription)); + downloadFinised = downloadService.isFinished(fileDescription); + } + } + + @Override + public void receiveResult(FileDescription fileDescription) { + this.fileDescription = fileDescription; + System.out.println("result received"); + } + +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/config/DisworkConfig.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/config/DisworkConfig.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/config/DisworkConfig.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,21 @@ +package org.nuiton.disworkfs.config; + +import java.io.File; +import java.util.Random; + +import org.nuiton.util.ApplicationConfig; + +public class DisworkConfig extends ApplicationConfig { + + public DisworkConfig() { + Random random = new Random(); + setDefaultOption("storage", "/tmp/disworkfs/storage" + random.nextInt()); + + // replication strategy... + } + + public File getStoragePath() { + return this.getOptionAsFile("storage"); + } + +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/messages/FileRequestMessage.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/messages/FileRequestMessage.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/messages/FileRequestMessage.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,22 @@ +package org.nuiton.disworkfs.messages; + +import java.io.Serializable; + +import org.nuiton.disworkfs.split.FileDescription; + +public class FileRequestMessage implements Serializable { + + /** + * + */ + private static final long serialVersionUID = 7208155655482510721L; + private FileDescription fileDescription; + + public FileRequestMessage(FileDescription fileDescription) { + this.fileDescription = fileDescription; + } + + public FileDescription getfileDescription() { + return this.fileDescription; + } +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/messages/FileTransferMessage.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/messages/FileTransferMessage.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/messages/FileTransferMessage.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,31 @@ +package org.nuiton.disworkfs.messages; + +import java.io.Serializable; + +import org.nuiton.disworkfs.split.FileChunk; +import org.nuiton.disworkfs.split.FileDescription; + +public class FileTransferMessage implements Serializable { + + /** + * + */ + private static final long serialVersionUID = -8759693577395988179L; + private FileChunk fileChunk; + private FileDescription fileDescription; + + public FileTransferMessage(FileChunk fileChunk, FileDescription fileDescription) { + this.fileChunk = fileChunk; + this.fileDescription = fileDescription; + } + + public FileChunk getFileChunk() { + return this.fileChunk; + } + + public FileDescription getFileDescrition() { + return this.fileDescription; + } + + +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/messages/LookUpMessage.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/messages/LookUpMessage.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/messages/LookUpMessage.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,20 @@ +package org.nuiton.disworkfs.messages; + +import java.io.Serializable; + +public class LookUpMessage implements Serializable { + + /** + * + */ + private static final long serialVersionUID = -1538199939808485195L; + private String fileName; + + public LookUpMessage(String fileName) { + this.fileName = fileName; + } + + public String getFileName() { + return this.fileName; + } +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/messages/LookUpResponseMessage.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/messages/LookUpResponseMessage.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/messages/LookUpResponseMessage.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,20 @@ +package org.nuiton.disworkfs.messages; + +import java.io.Serializable; + +import org.nuiton.disworkfs.split.FileDescription; + +public class LookUpResponseMessage implements Serializable { + + private static final long serialVersionUID = -8880348876968678210L; + + private FileDescription fileDescription; + + public LookUpResponseMessage(FileDescription fileDescription) { + this.fileDescription = fileDescription; + } + + public FileDescription getFileDescription() { + return this.fileDescription; + } +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/AbstractDisworkService.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/AbstractDisworkService.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/AbstractDisworkService.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,102 @@ +package org.nuiton.disworkfs.services; + +import java.io.Serializable; + +import org.jgroups.Address; +import org.jgroups.View; +import org.nuiton.disworkfs.config.DisworkConfig; +import org.nuiton.disworkfs.messages.FileRequestMessage; +import org.nuiton.disworkfs.messages.FileTransferMessage; +import org.nuiton.disworkfs.messages.LookUpMessage; +import org.nuiton.disworkfs.messages.LookUpResponseMessage; +import org.nuiton.disworkfs.transport.Message; +import org.nuiton.disworkfs.transport.Receiver; +import org.nuiton.disworkfs.transport.Transport; + +public abstract class AbstractDisworkService implements Receiver, Runnable { + + // protected JChannel jChannel; + + protected Transport transport; + + protected DisworkConfig disworkConfig; + + public AbstractDisworkService() {} + + /* + public AbstractDisworkService(JChannel jChannel) { + this.jChannel = jChannel; + } + */ + + public AbstractDisworkService(Transport transport) { + this.transport = transport; + } + + /* + public void receive(Message msg) { + Object obj = msg.getObject(); + if (obj instanceof LookUpMessage) { + receiveLookUpMessage(msg); + } else if (obj instanceof LookUpResponseMessage) { + receiveLookUpResponseMessage(msg); + } else if (obj instanceof FileRequestMessage) { + receiveFileRequestMessage(msg); + } else if (obj instanceof FileTransferMessage) { + receiveFileTransferMessage(msg); + } else { + System.out.println("unknow message received"); + System.out.println(obj); + } + } + */ + + @Override + public void receiveMessage(Message message) { + + Serializable obj = message.getContent(); + if (obj instanceof LookUpMessage) { + receiveLookUpMessage(message); + } else if (obj instanceof LookUpResponseMessage) { + receiveLookUpResponseMessage(message); + } else if (obj instanceof FileRequestMessage) { + receiveFileRequestMessage(message); + } else if (obj instanceof FileTransferMessage) { + receiveFileTransferMessage(message); + } else { + System.out.println("unknow message received"); + System.out.println(obj); + } + } + + public void receiveLookUpMessage(Message msg) {} + public void receiveLookUpResponseMessage(Message msg) {} + public void receiveFileRequestMessage(Message msg) {} + public void receiveFileTransferMessage(Message msg) {} + + public void viewAccepted(View new_view) { + } + + public void suspect(Address suspected_mbr) { + } + + /* + public void setJChannel(JChannel jChannel) { + this.jChannel = jChannel; + } + */ + + public void setTransport(Transport transport) { + this.transport = transport; + } + + @Override + public void run() { + } + + public void setDisworkConfig(DisworkConfig disworkConfig) { + this.disworkConfig = disworkConfig; + } + + +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/DownloadService.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/DownloadService.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/DownloadService.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,143 @@ +package org.nuiton.disworkfs.services; + +import java.io.File; +import java.io.IOException; +import java.util.HashMap; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.nuiton.disworkfs.DownloadObserver; +import org.nuiton.disworkfs.messages.FileRequestMessage; +import org.nuiton.disworkfs.messages.FileTransferMessage; +import org.nuiton.disworkfs.split.FileDescription; +import org.nuiton.disworkfs.split.SplitFileFromChunks; +import org.nuiton.disworkfs.transport.Message; + +public class DownloadService extends AbstractDisworkService { + + private Map<Long, SplitFileFromChunks> downloadsInProgress = new HashMap<Long, SplitFileFromChunks>(); + + private List<Long> finishedDownloads = new LinkedList<Long>(); + + private Map<Long, List<DownloadObserver>> downloadObservers = new HashMap<Long, List<DownloadObserver>>(); + + private static final Log log = LogFactory.getLog(DownloadService.class); + + @Override + public void receiveFileTransferMessage(Message message) { + + FileTransferMessage fileTransferMessage = (FileTransferMessage) message.getContent(); + + if (log.isDebugEnabled()) { + log.info("received file chunk " + + fileTransferMessage.getFileDescrition().getFileName() + + " chunk number " + + fileTransferMessage.getFileChunk().getBytesChunk().getChunkNumber() + + ")"); + } + + Long checkSum = fileTransferMessage.getFileDescrition().getFileCheckSum(); + + if (! downloadsInProgress.containsKey(checkSum)) { + // Start download + + log.info("first chunk received, initiate download"); + + SplitFileFromChunks newSplitFile = new SplitFileFromChunks(fileTransferMessage.getFileDescrition()); + + downloadsInProgress.put(checkSum, newSplitFile); + } + + // we have received a file chunk, let's add it + SplitFileFromChunks downloadingFile = downloadsInProgress.get(checkSum); + + downloadingFile.addChunk(fileTransferMessage.getFileChunk()); + + // maybe the download is complete + if (downloadingFile.isComplete()) { + // write file to the FS + + try { + File newFile = new File(disworkConfig.getStoragePath(), fileTransferMessage.getFileDescrition().getFileName()); + downloadingFile.writeToLocalFileSystem(newFile); + System.out.println("file written"); + } catch (IOException e) { + log.error("can't write file"); + } catch (Exception e) { + log.error("trying to write data with missing chunks"); + } + + finishedDownloads.add(checkSum); + downloadsInProgress.remove(checkSum); + + this.notifyAllDownloadObserversForFile(checkSum); + } + } + + public void registerObserver(FileDescription fileDescription, DownloadObserver downloadObserver) { + List<DownloadObserver> observersList = downloadObservers.get(fileDescription.getFileCheckSum()); + + if (observersList == null) { + // it's the first observer for this download ever, let's construct a list + observersList = new LinkedList<DownloadObserver>(); + } + observersList.add(downloadObserver); + downloadObservers.put(fileDescription.getFileCheckSum(), observersList); + } + + private void notifyAllDownloadObserversForFile(Long checksum) { + List<DownloadObserver> downloadObserversForThisFile = downloadObservers.get(checksum); + + // important check : maybe no observer for this file download + // so the list is null. + if (downloadObserversForThisFile != null) { + + for (DownloadObserver downloadObserver : downloadObserversForThisFile) { + downloadObserver.updateDownloadStatus(this); + } + } + } + + public boolean isFinished(FileDescription fileDescription) { + boolean finished = finishedDownloads.contains(fileDescription.getFileCheckSum()); + return finished; + } + + public double getProgess(FileDescription fileDescription) { + // FIXME this method never return intermediate values + double progress = 0.0; + if (isFinished(fileDescription)) { + progress = 1.0; + } else { + Long checksum = fileDescription.getFileCheckSum(); + + if (downloadsInProgress.containsKey(checksum)) { + // download is in progress + long actualSize = downloadsInProgress.get(checksum).getActualSize(); + long expectedSize = fileDescription.getTotalSize(); + progress = actualSize / expectedSize; + } else { + // download not started + progress = -1.0; + } + } + return progress; + } + + public void startDownload(FileDescription fileDescription, DownloadObserver downloadObserver) { + Message message = transport.newMulticastMessage(); + message.setContent(new FileRequestMessage(fileDescription)); + message.send(); + /* + Message message = new Message(null, null, new FileRequestMessage(fileDescription)); + jChannel.send(message); + */ + if (log.isDebugEnabled()) + log.info("sending file request for " + fileDescription.getFileName()); + registerObserver(fileDescription, downloadObserver); + + } +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/LookUpService.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/LookUpService.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/LookUpService.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,65 @@ +package org.nuiton.disworkfs.services; + + +import java.util.HashMap; +import java.util.Map; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.nuiton.disworkfs.LookUpObserver; +import org.nuiton.disworkfs.messages.LookUpMessage; +import org.nuiton.disworkfs.messages.LookUpResponseMessage; +import org.nuiton.disworkfs.transport.Message; + + + +public class LookUpService extends AbstractDisworkService { + + /** + * this map store, for each request, the LookUpObserver to notify when + * the LookUpResponse is received + */ + Map<String, LookUpObserver> requestToRequester = new HashMap<String, LookUpObserver>(); + + private static final Log log = LogFactory.getLog(LookUpService.class); + + /** + * + * @param fileName the name of the name + * @return the FileDescription or null of file have not been found + */ + public void lookForFileName(String fileName, LookUpObserver lookUpObserver) { + // TODO use a MessageDispatcher + // System.out.println("lookForFileName(" + fileName + ")"); + + LookUpMessage lookUpMessage = new LookUpMessage(fileName); + + /* + Message message = new Message(null, null, lookUpMessage); + */ + Message message = transport.newMulticastMessage(); + message.setContent(lookUpMessage); + message.send(); + log.info("look-up message sent for " + fileName); + + requestToRequester.put(fileName, lookUpObserver); + // System.out.println("expected == " + fileName); + } + + @Override + public void receiveLookUpResponseMessage(Message message) { + + LookUpResponseMessage lookUpResponse = (LookUpResponseMessage) message.getContent(); + + String requestedFileName = lookUpResponse.getFileDescription().getFileName(); + + // System.out.println("actual == " + requestedFileName); + if (requestToRequester.containsKey(requestedFileName)) { + LookUpObserver requester = requestToRequester.get(requestedFileName); + requester.receiveResult(lookUpResponse.getFileDescription()); + requestToRequester.remove(requestedFileName); + } + + } + +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/UploadService.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/UploadService.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/services/UploadService.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,114 @@ +package org.nuiton.disworkfs.services; + +import java.io.File; +import java.io.IOException; +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.nuiton.disworkfs.messages.FileRequestMessage; +import org.nuiton.disworkfs.messages.FileTransferMessage; +import org.nuiton.disworkfs.messages.LookUpMessage; +import org.nuiton.disworkfs.messages.LookUpResponseMessage; +import org.nuiton.disworkfs.split.FileChunk; +import org.nuiton.disworkfs.split.FileDescription; +import org.nuiton.disworkfs.split.SplitFileFromLocalFile; +import org.nuiton.disworkfs.transport.Message; + +/** + * When receiving a FileRequest, sending the file + * When receiving a LookUp, sending a LookUpResponse if owning file on local FS + */ +public class UploadService extends AbstractDisworkService { + + private static final Log log = LogFactory.getLog(UploadService.class); + + @Override + public void receiveFileRequestMessage(Message message) { + FileRequestMessage fileRequestMessage = (FileRequestMessage) message.getContent(); + + File file = new File(disworkConfig.getStoragePath(), fileRequestMessage.getfileDescription().getFileName()); + + if (log.isDebugEnabled()) + log.info("file request message received for " + fileRequestMessage.getfileDescription().getFileName()); + + if (file.exists()) { + + if (log.isDebugEnabled()) + log.info("file found in path " + file.getAbsolutePath()); + + SplitFileFromLocalFile splitFileFromLocalFile = new SplitFileFromLocalFile(file); + + try { + FileDescription fileDescription = splitFileFromLocalFile.getFileDescription(); + fileDescription.setFileName(fileRequestMessage.getfileDescription().getFileName()); + + List<FileChunk> chunks = splitFileFromLocalFile.getAllChunks(); + + for (FileChunk fileChunk : chunks) { + Message reply = message.newReply(); + + FileTransferMessage fileTransferMessage = new FileTransferMessage(fileChunk, fileDescription); + + reply.setContent(fileTransferMessage); + + if (log.isDebugEnabled()) + log.info("sending chunk " + fileTransferMessage.getFileChunk().getBytesChunk().getChunkNumber()); + + reply.send(); + + } + + if (log.isDebugEnabled()) + log.info("all chunks sent"); + + } catch (IOException e) { + log.error("can't read file " + file.getAbsolutePath()); + } + } else { + if (log.isDebugEnabled()) + log.info("file not found in path " + file.getAbsolutePath() + " don't send response"); + } + + } + + @Override + public void receiveLookUpMessage(Message message) { + LookUpMessage lookUpMessage = (LookUpMessage) message.getContent(); + + if (log.isDebugEnabled()) + log.info("lookup message received : looking for file " + lookUpMessage.getFileName()); + + File file = new File(disworkConfig.getStoragePath(), lookUpMessage.getFileName()); + + + if (file.exists()) { + if (log.isDebugEnabled()) + log.info(lookUpMessage.getFileName() + " file found at " + file.getAbsolutePath()); + + SplitFileFromLocalFile splitFileFromLocalFile = new SplitFileFromLocalFile(file); + + try { + + Message reply = message.newReply(); + + FileDescription fileDescription = splitFileFromLocalFile.getFileDescription(); + + fileDescription.setFileName(lookUpMessage.getFileName()); + + reply.setContent(new LookUpResponseMessage(fileDescription)); + + if (log.isDebugEnabled()) + log.info("sending lookUpResponse response for " + lookUpMessage.getFileName()); + + reply.send(); + } catch (IOException e) { + log.error("can't read file " + file.getAbsolutePath() + " aborting file transfer"); + } + } else { + if (log.isDebugEnabled()) + log.info(lookUpMessage.getFileName() + " file not found at " + file.getAbsolutePath() + " no response sent"); + } + + } +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/BytesChunk.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/BytesChunk.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/BytesChunk.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,50 @@ +package org.nuiton.disworkfs.split; + +import java.io.Serializable; + +// TODO really implement Serializable +public class BytesChunk implements Serializable { + + /** + * + */ + private static final long serialVersionUID = 2562740291339559768L; + + // FIXME dangerous : not final ! + public static int MAX_CHUNK_SIZE = 1024; + + private byte[] data = new byte[MAX_CHUNK_SIZE]; + + private int chunkNumber; + + private int chunkSize; + + public BytesChunk(byte[] data) { + this.setData(data); + } + + public byte[] getData() { + return this.data; + } + + public void setData(byte[] data) { + this.data = data; + } + + public int getChunkNumber() { + return this.chunkNumber; + } + + public void setChunkNumber(int chunkNumber) { + this.chunkNumber = chunkNumber; + } + + public int getChunkSize() { + return this.chunkSize; + } + + public void setChunkSize(int chunkSize) { + this.chunkSize = chunkSize; + } + +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/FileChunk.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/FileChunk.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/FileChunk.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,21 @@ +package org.nuiton.disworkfs.split; + +import java.io.Serializable; + +public class FileChunk implements Serializable { + + /** + * + */ + private static final long serialVersionUID = 7668123461277672243L; + private BytesChunk bytesChunk; + + public FileChunk(BytesChunk bytesChunk) { + this.bytesChunk = bytesChunk; + } + + public BytesChunk getBytesChunk() { + return bytesChunk; + } + +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/FileDescription.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/FileDescription.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/FileDescription.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,45 @@ +package org.nuiton.disworkfs.split; + +import java.io.Serializable; + +public class FileDescription implements Serializable { + + private static final long serialVersionUID = 1809101246917954092L; + private String fileName; + private long totalSize; + private long fileCheckSum; + + + public FileDescription(String fileName, long totalSize, long fileCheckSum) { + super(); + this.fileName = fileName; + this.totalSize = totalSize; + this.fileCheckSum = fileCheckSum; + } + + + public String getFileName() { + return fileName; + } + + + public long getTotalSize() { + return totalSize; + } + + + public long getFileCheckSum() { + return fileCheckSum; + } + + + public int getNumberOfChunks() { + return SplittedBytes.numberOfChunksNeededtoStore(this.totalSize); + } + + public void setFileName(String fileName) { + this.fileName = fileName; + } + + +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitFileFromChunks.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitFileFromChunks.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitFileFromChunks.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,80 @@ +package org.nuiton.disworkfs.split; + +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.zip.CRC32; + +import org.apache.commons.io.FileUtils; +import org.apache.commons.lang.NotImplementedException; + +public class SplitFileFromChunks { + + private FileDescription fileDescription; + + private List<FileChunk> fileChunks; + + + public SplitFileFromChunks(FileDescription fileDescription) { + this.fileDescription = fileDescription; + fileChunks = new ArrayList<FileChunk>(); + } + + public void addChunk(FileChunk fileChunk) { + fileChunks.add(fileChunk); + } + + public boolean isComplete() { + return fileChunks.size() == fileDescription.getNumberOfChunks(); + } + + public long getActualSize() { + int actualSize = 0; + for (FileChunk fileChunk : fileChunks) { + actualSize += fileChunk.getBytesChunk().getChunkSize(); + } + return actualSize; + } + + public List<Integer> getMissingChunksIds() { + throw new NotImplementedException(); + } + + + public boolean writeToLocalFileSystem() throws IOException, Exception { + File newFile = new File(fileDescription.getFileName()); + return writeToLocalFileSystem(newFile); + } + + /** + * + * @return the result of the check integrity + * @throws IOException + * @throws Exception + */ + public boolean writeToLocalFileSystem(File newFile) throws IOException, Exception { + + if (fileChunks == null) { + throw new Exception("No data to write"); + } + + + long expectedSize = fileDescription.getTotalSize(); + + SplittedBytes splittedBytes = new SplittedBytes(expectedSize); + + for (FileChunk fileChunk : fileChunks) { + splittedBytes.addChunk(fileChunk.getBytesChunk()); + } + + FileUtils.writeByteArrayToFile(newFile, splittedBytes.getBytesFromChunks()); + + // checking integrity + long expectedChecksum = fileDescription.getFileCheckSum(); + long actualCheckSum = FileUtils.checksum(newFile, new CRC32()).getValue(); + + return actualCheckSum == expectedChecksum; + } + +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitFileFromLocalFile.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitFileFromLocalFile.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitFileFromLocalFile.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,63 @@ +package org.nuiton.disworkfs.split; + +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.zip.CRC32; + +import org.apache.commons.io.FileUtils; + +public class SplitFileFromLocalFile { + + private File localFile; + + public SplitFileFromLocalFile(File file) { + localFile = file; + } + + public FileDescription getFileDescription() throws IOException { + System.out.println("returning a file description with " + localFile.getPath()); + String localFileName = localFile.getPath(); + long localFileTotalSize = localFile.length(); + long localFileCheckSum = FileUtils.checksum(localFile, new CRC32()).getValue(); + return new FileDescription(localFileName, localFileTotalSize, localFileCheckSum); + } + + public List<FileChunk> getAllChunks() throws IOException { + byte[] data = FileUtils.readFileToByteArray(localFile); + + // Creating the file chunks + SplittedBytes splittedBytes = new SplittedBytes(); + splittedBytes.setChunksFromBytes(data); + + List<FileChunk> fileChunks = new ArrayList<FileChunk>(); + for(BytesChunk bytesChunk : splittedBytes.getChunks()) { + fileChunks.add(new FileChunk(bytesChunk)); + } + return fileChunks; + } + + public List<FileChunk> getSomeChunks(int[] indexes) throws IOException { + // TODO implement + List<FileChunk> allChunks = this.getAllChunks(); + List<FileChunk> someChunks = new ArrayList<FileChunk>(); + for (int index : indexes) { + someChunks.add(allChunks.get(index)); + } + return someChunks; + } + + public FileChunk getChunk(int index) throws IOException { + int singletonIndex[] = { index }; + + return this.getSomeChunks(singletonIndex).get(0); + + } + + public int getNumberOfChunks() throws IOException { + return this.getFileDescription().getNumberOfChunks(); + } + + +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitedFile.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitedFile.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplitedFile.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,3 @@ +package org.nuiton.disworkfs.split; + + Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplittedBytes.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplittedBytes.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/split/SplittedBytes.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,190 @@ +package org.nuiton.disworkfs.split; + +import java.util.ArrayList; +import java.util.List; + + +/** + * This class permit a great array of bytes (a file ?) to be split in multiple + * chunks. This class offers methods to split and unsplit an array of bytes. + * + * There is two use for this class : + * <ul> + * <li>Get a collection of chunks, given a single big array of bytes</li> + * <li>Get a single big array of bytes, given all the chunks</li> + * </ul> + * + * In the first case, you should use the SplittedBytes() constructor. Then, use + * the setChunksFromBytes method. Finally, use getChunks() to get all the + * generated chunks. + * + * In the second case, you can re-construct, step by step, adding chunks by + * chunks. First, use SplittedBytes(int expectedSize) to construct by providing + * the size of the array. Then, add chunks by chunks using addChunk. You don't + * need to provide the chunks in a particular order. It's useful if you're + * obtaining your chunks in an unknown order. You just have to provide all the + * chunks (until isComplete is true). Finally, get the final full array with + * getBytesFromChunks(). + * + * If you are missing some chunks, you can know who of them is missing with + * missingChinks() + * + * @author bleny + */ + + +public class SplittedBytes { + + private List<BytesChunk> chunks; + private long totalSize; + + public static int numberOfChunksNeededtoStore(long numberOfBytes) { + long longNumberOfChunks = numberOfBytes / BytesChunk.MAX_CHUNK_SIZE; + + // FIXME + int numberOfChunks = (int) longNumberOfChunks; + + if (numberOfBytes > numberOfChunks * BytesChunk.MAX_CHUNK_SIZE) { + numberOfChunks += 1; + } + return numberOfChunks; + } + + /** + * Use this constructor if you're building a byte-array from chunks. You + * have to know what will be the exact size of final array : this is needed + * to make isComplete working The argument is a number of bytes, not a + * number of chunks ! + * + * @param expectedSize + * the size of the array of bytes you will get + */ + public SplittedBytes(long expectedSize) { + this.totalSize = expectedSize; + + // create a list with one place for each chunks + List<BytesChunk> emptyChunkList = new ArrayList<BytesChunk>( + numberOfChunksNeededtoStore(expectedSize)); + for (int chunkNumber = 0; chunkNumber < numberOfChunksNeededtoStore(expectedSize); ++chunkNumber) { + emptyChunkList.add(null); + } + this.chunks = emptyChunkList; + } + + public SplittedBytes() { + this.chunks = new ArrayList<BytesChunk>(); + } + + public long getTotalSize() { + return this.totalSize; + } + + public void setChunksFromBytes(byte[] data) { + int numberOfChunks = data.length / BytesChunk.MAX_CHUNK_SIZE; + if (data.length > numberOfChunks * BytesChunk.MAX_CHUNK_SIZE) { + numberOfChunks += 1; + } + + // System.out.println("numberOfChunks = " + numberOfChunks); + + for (int currentChunk = 0; currentChunk < numberOfChunks; ++currentChunk) { + + // System.out.println("currentChunk = " + currentChunk); + + int numberOfLastBytes = BytesChunk.MAX_CHUNK_SIZE; + + if (data.length - (currentChunk * BytesChunk.MAX_CHUNK_SIZE) < BytesChunk.MAX_CHUNK_SIZE) { + // System.out.println("last chunk"); + numberOfLastBytes = data.length - currentChunk + * BytesChunk.MAX_CHUNK_SIZE; + } + + // System.out.println("reading " + numberOfLastBytes + " bytes"); + + byte[] chunkData = new byte[BytesChunk.MAX_CHUNK_SIZE]; + + System.arraycopy(data, currentChunk * BytesChunk.MAX_CHUNK_SIZE, + chunkData, 0, numberOfLastBytes); + + BytesChunk bytesChunk = new BytesChunk(chunkData); + bytesChunk.setChunkNumber(currentChunk); + bytesChunk.setChunkSize(numberOfLastBytes); + this.chunks.add(bytesChunk); + totalSize += numberOfLastBytes; + + /* + for (int i = 0; i < chunkData.length; i++) { + System.out.println("chunkData[" + i + "] = " + chunkData[i]); + } + */ + } + } + + public List<BytesChunk> getChunks() { + return this.chunks; + } + + public byte[] getBytesFromChunks() throws Exception { + + if (!this.isComplete()) { + // TODO use a specific exception ? + throw new Exception("Data are incomplete : chunks are missing"); + } + + byte[] data = new byte[(int) totalSize]; + int bytesReads = 0; + totalSize = 0; + for (BytesChunk fileChunk : this.getChunks()) { + // System.out.println("reading chunks n° " + + // fileChunk.getChunkNumber()); + // System.out.println("reading " + fileChunk.getChunkSize()); + + // System.out.println("writing at "+fileChunk.getChunkSize()+" bytes to data["+bytesReads+"]" + // + fileChunk.getData()); + System.arraycopy(fileChunk.getData(), 0, data, bytesReads, + fileChunk.getChunkSize()); + + /* + for (int i = 0; i < fileChunk.getChunkSize(); i++) { + byte[] tempdata = fileChunk.getData(); + System.out.println("tempdata[" + i + "] = " + tempdata[i]); + } + */ + + bytesReads += fileChunk.getChunkSize(); + totalSize += bytesReads; + } + + return data; + } + + public void addChunk(BytesChunk chunk) { + this.getChunks().set(chunk.getChunkNumber(), chunk); + } + + public boolean isComplete() { + // return this.missingChunks().isEmpty(); + return totalSize == this.getActualSize(); + } + + public int getActualSize() { + int result = 0; + for (BytesChunk bytesChunk : chunks) { + if (bytesChunk != null) { + result += bytesChunk.getChunkSize(); + } + } + return result; + } + + public List<Integer> missingChunks() { + List<Integer> missingChunks = new ArrayList<Integer>(); + for (int chunkNumber = 0; chunkNumber < this.getChunks().size(); ++chunkNumber) { + if (this.getChunks().get(chunkNumber) == null) { + missingChunks.add(chunkNumber); + } + } + return missingChunks; + } + +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/Address.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/Address.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/Address.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,7 @@ +package org.nuiton.disworkfs.transport; + +import java.io.Serializable; + +public abstract class Address implements Serializable { + +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/Message.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/Message.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/Message.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,39 @@ +package org.nuiton.disworkfs.transport; + +import java.io.Serializable; + + +/** + * Use Transport as a factory to create new instances of this class + * @author bleny + * + */ +public abstract class Message { + + protected Transport transport; + + /** + * contructor is not available as public. Use {@link Transport#newEmptyMessage()} + * if you want to create a new message + * @param transport + */ + protected Message(Transport transport) { + // dependency injection done by the newEmptyMessage factory + this.transport = transport; + } + + public abstract void setSource(Address source); + public abstract void setDestination(Address destination); + public abstract void setContent(Serializable content); + + public abstract Address getSource(); + public abstract Address getDestination(); + public abstract Serializable getContent(); + + public void send() { + this.transport.send(this); + } + + public abstract Message newReply(); + +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/Receiver.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/Receiver.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/Receiver.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,7 @@ +package org.nuiton.disworkfs.transport; + +public interface Receiver { + + public void receiveMessage(Message message); + +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/Transport.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/Transport.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/Transport.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,43 @@ +package org.nuiton.disworkfs.transport; + +public interface Transport { + + /** + * @return true if the message have been sent without error + */ + public boolean send(Message message); + + /** + * Factory method + * @return a new empty message, ready to be sent + */ + public Message newEmptyMessage(); + + /** + * Factory method + * @return a new empty message, ready to be sent to all nodes + */ + public Message newMulticastMessage(); + + /** + * Use this method to set the receiver for this transport + * every message received by this transport will be sent + * to this object + * @param receiver + */ + public void setReceiver(Receiver receiver); + + /** + * @return an address you should use in {@link Message#setSource(Address)} + */ + public Address getLocalAddress(); + + /** + * if you want to send a message to all nodes, use this method + * to get the address you will use as destination in your + * message + * @return an address you can use in Message.setDestination() + */ + public Address getMulticastAddress(); + +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/jgroups/JGroupsAddress.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/jgroups/JGroupsAddress.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/jgroups/JGroupsAddress.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,19 @@ +package org.nuiton.disworkfs.transport.jgroups; + +import org.nuiton.disworkfs.transport.Address; + +public class JGroupsAddress extends Address { + + private static final long serialVersionUID = 8851527317522260037L; + + org.jgroups.Address address; + + public JGroupsAddress(org.jgroups.Address address) { + this.address = address; + } + + public org.jgroups.Address getAddress() { + return address; + } + +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/jgroups/JGroupsMessage.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/jgroups/JGroupsMessage.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/jgroups/JGroupsMessage.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,81 @@ +package org.nuiton.disworkfs.transport.jgroups; + +import java.io.Serializable; + +import org.nuiton.disworkfs.transport.Address; +import org.nuiton.disworkfs.transport.Message; +import org.nuiton.disworkfs.transport.Transport; + +public class JGroupsMessage extends Message implements Serializable { + + /** + * + */ + private static final long serialVersionUID = 6239305387929159827L; + protected org.jgroups.Message message; + + protected JGroupsMessage(Transport transport) { + super(transport); + this.message = new org.jgroups.Message(); + } + + protected JGroupsMessage(Transport transport, org.jgroups.Message message) { + this(transport); + this.message = message; + } + + protected JGroupsMessage(Transport transport, + org.jgroups.Address source, + org.jgroups.Address destination, + Serializable content) { + this(transport); + this.message = new org.jgroups.Message(source, destination, content); + } + + @Override + public Serializable getContent() { + return (Serializable) this.message.getObject(); + } + + @Override + public Address getDestination() { + JGroupsAddress destination = new JGroupsAddress(message.getDest()); + return destination; + } + + @Override + public Address getSource() { + JGroupsAddress destination = new JGroupsAddress(message.getSrc()); + return destination; + } + + @Override + public void setContent(Serializable content) { + this.message.setObject(content); + } + + @Override + public void setDestination(Address destination) { + org.jgroups.Address dest = ((JGroupsAddress) destination).getAddress(); + this.message.setDest(dest); + } + + @Override + public void setSource(Address source) { + org.jgroups.Address src = ((JGroupsAddress) source).getAddress(); + this.message.setDest(src); + } + + public org.jgroups.Message getMessage() { + return this.message; + } + + @Override + public Message newReply() { + Message reply = new JGroupsMessage(this.transport); + reply.setDestination(this.getSource()); + reply.setSource(this.getDestination()); + return reply; + } + +} Added: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/jgroups/JGroupsTransport.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/jgroups/JGroupsTransport.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/transport/jgroups/JGroupsTransport.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,110 @@ +package org.nuiton.disworkfs.transport.jgroups; + +import java.io.Serializable; + +import org.jgroups.Channel; +import org.jgroups.ChannelClosedException; +import org.jgroups.ChannelException; +import org.jgroups.ChannelNotConnectedException; +import org.jgroups.JChannel; +import org.jgroups.View; +import org.nuiton.disworkfs.config.DisworkConfig; +import org.nuiton.disworkfs.services.AbstractDisworkService; +import org.nuiton.disworkfs.transport.Address; +import org.nuiton.disworkfs.transport.Message; +import org.nuiton.disworkfs.transport.Receiver; +import org.nuiton.disworkfs.transport.Transport; + +public class JGroupsTransport implements Transport { + + protected DisworkConfig disworkConfig; + protected JChannel jChannel; + protected Receiver receiver; + + /** + * This class is just a delegator. Every message received on the jChannel, + * will be sent to the Receiver who subscribed to transport + * @author bleny + */ + private class MyReceiver extends org.jgroups.ExtendedReceiverAdapter { + + private Transport tranport; + + public MyReceiver(Transport tranport) { + this.tranport = tranport; + } + + public void receive(org.jgroups.Message msg) { + + Message message = tranport.newEmptyMessage(); + + message.setSource(new JGroupsAddress(msg.getSrc())); + message.setSource(new JGroupsAddress(msg.getDest())); + message.setContent((Serializable) msg.getObject()); + + receiver.receiveMessage(message); + } + + } + + + public JGroupsTransport(DisworkConfig disworkConfig) { + this.disworkConfig = disworkConfig; + try { + jChannel = new JChannel("udp.xml"); + jChannel.connect("MonPremierGroupe"); + + // don't receive messages sent by myself + jChannel.setOpt(Channel.LOCAL, new Boolean(false)); + + MyReceiver myReceiver = new MyReceiver(this); + jChannel.setReceiver(myReceiver); + + } catch (ChannelException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + } + + public void setReceiver(Receiver receiver) { + this.receiver = receiver; + } + + public Address getMulticastAddress() { + // null is the multicast destination in JGroups + return new JGroupsAddress(null); + } + + public boolean send(Message message) { + org.jgroups.Message msg = ((JGroupsMessage) message).getMessage(); + try { + jChannel.send(msg); + } catch (ChannelNotConnectedException e) { + // TODO log + return false; + } catch (ChannelClosedException e) { + // TODO log + return false; + } + return true; + } + + public Message newEmptyMessage() { + Message message = new JGroupsMessage(this, new org.jgroups.Message()); + message.setSource(this.getLocalAddress()); + return message; + } + + public Message newMulticastMessage() { + Message message = this.newEmptyMessage(); + message.setDestination(this.getMulticastAddress()); + return message; + } + + @Override + public Address getLocalAddress() { + // null is the local address in JGroups + return new JGroupsAddress(jChannel.getAddress()); + } + +} Added: trunk/diswork-fs/src/main/resources/log4j.properties =================================================================== --- trunk/diswork-fs/src/main/resources/log4j.properties (rev 0) +++ trunk/diswork-fs/src/main/resources/log4j.properties 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,8 @@ +# Global logging configuration +log4j.rootLogger=WARN, stdout +# Console output... +log4j.appender.stdout=org.apache.log4j.ConsoleAppender +log4j.appender.stdout.layout=org.apache.log4j.PatternLayout +log4j.appender.stdout.layout.ConversionPattern=%d %5p [%t] (%F:%L) %M - %m%n +# package level +log4j.logger.org.nuiton.disworkfs=DEBUG Added: trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/AbstractSplitFileTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/AbstractSplitFileTest.java (rev 0) +++ trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/AbstractSplitFileTest.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,56 @@ +package org.nuiton.disworkfs; + +import java.io.File; +import java.util.Random; + +import org.apache.commons.io.FileUtils; +import org.junit.After; +import org.junit.Before; + +/** + * This classes is for test purpose. It creates, before all tests + * a random file with a given size in a temp directory + */ +public abstract class AbstractSplitFileTest { + + static protected Random random = new Random(); + + /** + * a place to store files for the test + * it's a subdirectory of the OS temp dir + * e.g. under linux /tmp/disworkfs/tests/ + */ + static protected String tempDirectoryPath = System.getProperty("java.io.tmpdir", ".") + "/disworkfs/tests"; + + /** + * We will create a file at this path for test purpose + */ + static protected String randomFilePath = tempDirectoryPath + "/randomfile"; + + /** + * The file will have this fixed size + */ + static protected int randomFileSize = 3500; + + @Before + public void setUp() throws Exception { + File tempDirectory = new File(tempDirectoryPath); + tempDirectory.mkdir(); + + // creating random data for the file + byte[] randomBytes = new byte[randomFileSize]; + random.nextBytes(randomBytes); + + // dumping random data into the file + File randomFile = new File(randomFilePath); + FileUtils.writeByteArrayToFile(randomFile, randomBytes); + } + + @After + public void tearDown() throws Exception { + // cleaning + new File(randomFilePath).delete(); + new File(tempDirectoryPath).delete(); + } + +} Added: trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/AllwaysReplyToLookUpTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/AllwaysReplyToLookUpTest.java (rev 0) +++ trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/AllwaysReplyToLookUpTest.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,34 @@ +package org.nuiton.disworkfs; + +import static org.mockito.Mockito.mock; + +import org.jgroups.JChannel; +import org.junit.Test; +import org.mockito.Mock; + +public class AllwaysReplyToLookUpTest extends AbstractSplitFileTest { + + @Mock JChannel mockedJChannel = mock(JChannel.class); + + + @Test + public void testReceive() { + // FIXME can't mock Message because getObject() is final + /* + + AllwaysReplyToLookUp allwaysReplyToLookUp = new AllwaysReplyToLookUp(mockedJChannel); + + Message mockedMessage = mock(Message.class); + // when(mockedMessage.getObject()).thenReturn(new LookUpMessage("randomFilePath")); + + + allwaysReplyToLookUp.receive(mockedMessage); + + + // verify(mockedJChannel, times(4)).send(mockedMessage); + + // when(message.getObject()).thenReturn(new LookUpMessage("randomFilePath")); + + */ + } +} Added: trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/DistributedFileSystemTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/DistributedFileSystemTest.java (rev 0) +++ trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/DistributedFileSystemTest.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,92 @@ +package org.nuiton.disworkfs; + +import java.io.File; +import java.util.Random; + +import org.apache.commons.io.FileUtils; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.nuiton.disworkfs.config.DisworkConfig; +import org.nuiton.util.FileUtil; + +public class DistributedFileSystemTest { + + /** + * a place to store files for the test + * it's a subdirectory of the OS temp dir + * e.g. under linux /tmp/disworkfs/tests/ + */ + static protected String tempDirectoryPath = System.getProperty("java.io.tmpdir", ".") + "/disworkfs/tests"; + + /** + * We will create a file at this path for test purpose + */ + static protected String randomFilePath = tempDirectoryPath + "/randomfile"; + + /** + * The file will have this fixed size + */ + static protected int randomFileSize = 10 * 1000 * 1000; + + static protected String storagePath1; + static protected String storagePath2; + + static protected DisworkFileSystem disworkFileSystem1; + static protected DisworkFileSystem disworkFileSystem2; + + @Before + public void setUp() throws Exception { + File tempDirectory = new File(tempDirectoryPath); + tempDirectory.mkdir(); + + + File storage1 = new File(tempDirectory, "storage1"); + storage1.mkdir(); + storagePath1 = storage1.getAbsolutePath(); + File storage2 = new File(tempDirectory, "storage2"); + storage2.mkdir(); + storagePath2 = storage2.getAbsolutePath(); + + Random random = new Random(); + // creating random data for the file + byte[] randomBytes = new byte[randomFileSize]; + random.nextBytes(randomBytes); + + // dumping random data into the file + File randomFile = new File(randomFilePath); + FileUtils.writeByteArrayToFile(randomFile, randomBytes); + + + DisworkConfig disworkConfig1 = new DisworkConfig(); + disworkConfig1.setOption("storage", storagePath1); + disworkFileSystem1 = new DisworkFileSystem(disworkConfig1); + + DisworkConfig disworkConfig2 = new DisworkConfig(); + disworkConfig2.setOption("storage", storagePath2); + disworkFileSystem2 = new DisworkFileSystem(disworkConfig2); + } + + @After + public void tearDown() throws Exception { + // cleaning + FileUtil.deleteRecursively(tempDirectoryPath); + } +/* + @Test + public void testWrite() throws Exception { + + disworkFileSystem1.write("monfichier", new File(randomFilePath)); + + } +*/ + @Test + public void testRead() throws Exception { + + disworkFileSystem1.write("monfichier", new File(randomFilePath)); + + disworkFileSystem2.read("monfichier"); + + } + +} Added: trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplitFileFromChunksTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplitFileFromChunksTest.java (rev 0) +++ trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplitFileFromChunksTest.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,72 @@ +package org.nuiton.disworkfs.split; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; + +import org.apache.commons.io.IOUtils; +import org.junit.Test; +import org.nuiton.disworkfs.AbstractSplitFileTest; + + +public class SplitFileFromChunksTest extends AbstractSplitFileTest { + + @Test + public void simpleCopy() { + + try { + File randomFile = new File(randomFilePath); + + // let's take a file + SplitFileFromLocalFile splitFile = new SplitFileFromLocalFile(randomFile); + + // let's create a second file + FileDescription fileDescription = splitFile.getFileDescription(); + String splitedFileCopyPath = fileDescription.getFileName() + "_copy"; + fileDescription.setFileName(splitedFileCopyPath); + + SplitFileFromChunks splitFileCopy = new SplitFileFromChunks(fileDescription); + + // here is the simple copy from fist file to second file + for (FileChunk fileChunk : splitFile.getAllChunks()) { + assertFalse(splitFileCopy.isComplete()); + assertTrue(splitFileCopy.getActualSize() < fileDescription.getTotalSize()); + splitFileCopy.addChunk(fileChunk); + } + + assertEquals(splitFileCopy.getActualSize(), fileDescription.getTotalSize()); + assertTrue(splitFileCopy.isComplete()); + // write the copy to the FS + boolean result = splitFileCopy.writeToLocalFileSystem(); + assertTrue("checking copy integrity", result); + + // compare the original and the copy byte by byte + try { + boolean actualContentEquality = IOUtils.contentEquals( + new FileInputStream(new File(randomFilePath)), + new FileInputStream(new File(splitedFileCopyPath)) + ); + assertTrue("file and copy content should be the same", actualContentEquality); + } catch (IOException e) { + fail("one or both files are not readable"); + e.printStackTrace(); + } + + // delete the copy + new File(splitedFileCopyPath).delete(); + + + } catch (IOException e) { + e.printStackTrace(); + } catch (Exception e) { + e.printStackTrace(); + } + + } + +} Added: trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplitFileFromLocalFileTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplitFileFromLocalFileTest.java (rev 0) +++ trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplitFileFromLocalFileTest.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,36 @@ +package org.nuiton.disworkfs.split; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.io.File; +import java.io.IOException; + +import org.junit.Test; +import org.nuiton.disworkfs.AbstractSplitFileTest; + +public class SplitFileFromLocalFileTest extends AbstractSplitFileTest { + + @Test + public void testReadFromLocalFileSytem() { + try { + File randomFile = new File(randomFilePath); + + SplitFileFromLocalFile splitFileFromLocalFile = new SplitFileFromLocalFile(randomFile); + + FileDescription fileDescription = splitFileFromLocalFile.getFileDescription(); + assertTrue(fileDescription.getTotalSize() == randomFileSize); + assertTrue(fileDescription.getFileCheckSum() != 0); + assertEquals(SplittedBytes.numberOfChunksNeededtoStore(randomFileSize), fileDescription.getNumberOfChunks()); + assertEquals(fileDescription.getFileName(), randomFilePath); + + splitFileFromLocalFile.getAllChunks(); + + } catch (IOException e) { + fail(); + e.printStackTrace(); + } + } + +} Added: trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplitedFileTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplitedFileTest.java (rev 0) +++ trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplitedFileTest.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,115 @@ +/*package org.nuiton.disworkfs.split; + + +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.util.Random; + +import org.apache.commons.io.FileUtils; +import org.apache.commons.io.IOUtils; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.nuiton.disworkfs.split.SplitedFile; + +public class SplitedFileTest { + + static Random random = new Random(); + + /** + * a place to store files for the test + * it's a subdirectory of the OS temp dir + * e.g. under linux /tmp/disworkfs/tests/ + + static String tempDirectoryPath = System.getProperty("java.io.tmpdir", ".") + "/disworkfs/tests"; + + *//** + * We will create a file at this path for test purpose + *//* + static String randomFilePath = tempDirectoryPath + "/randomfile"; + + *//** + * The file will have this fixed size + *//* + static int randomFileSize = 1 * 1000 * 1000; // 1 MB + + @Before + public void setUp() throws Exception { + File tempDirectory = new File(tempDirectoryPath); + tempDirectory.mkdir(); + + // creating random data for the file + byte[] randomBytes = new byte[randomFileSize]; + random.nextBytes(randomBytes); + + // dumping random data into the file + File randomFile = new File(randomFilePath); + FileUtils.writeByteArrayToFile(randomFile, randomBytes); + } + + @Test + public void testReadFromLocalFileSytem() { + try { + SplitedFile splitedFile = new SplitedFile(randomFilePath); + splitedFile.readFromLocalFileSytem(); + } catch (IOException e) { + fail(); + e.printStackTrace(); + } + } + + @Test + public void simpleCopy() { + + String splitedFileCopyPath = randomFilePath + "_copy"; + try { + // let's take a file + SplitedFile splitedFile = new SplitedFile(randomFilePath); + splitedFile.readFromLocalFileSytem(); + + + // let's create a second file + SplitedFile splitedFileCopy = new SplitedFile(splitedFileCopyPath); + + // here is the simple copy from fist file to second file + splitedFileCopy.setFileChunks(splitedFile.getFileChunks()); + + // write the copy to the FS + splitedFileCopy.writeToLocalFileSystem(); + + // compare the original and the copy byte by byte + try { + boolean actualContentEquality = IOUtils.contentEquals( + new FileInputStream(new File(randomFilePath)), + new FileInputStream(new File(splitedFileCopyPath)) + ); + assertTrue("file and copy content should be the same", actualContentEquality); + } catch (IOException e) { + fail("one or both files are not readable"); + e.printStackTrace(); + } + + // delete the copy + new File(splitedFileCopyPath).delete(); + } catch (IOException e) { + fail(); + e.printStackTrace(); + } catch (Exception e) { + fail(); + e.printStackTrace(); + } + } + + @After + public void tearDown() throws Exception { + // cleaning + new File(randomFilePath).delete(); + new File(tempDirectoryPath).delete(); + } + +} +*/ \ No newline at end of file Added: trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplittedBytesTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplittedBytesTest.java (rev 0) +++ trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/split/SplittedBytesTest.java 2010-05-04 16:58:34 UTC (rev 19) @@ -0,0 +1,117 @@ +package org.nuiton.disworkfs.split; + + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import org.junit.Before; +import org.junit.Test; + +public class SplittedBytesTest { + static final byte[] rawData = { 1, 2, 3, 4, 5, 6, 7, 8 }; + + static SplittedBytes splittedBytes = new SplittedBytes(); + + @Before + public void setUp() throws Exception { + BytesChunk.MAX_CHUNK_SIZE = 3; + } + + + /** + * this test show that bytes to chunks and chunks + * to bytes conversions are reflexive + * @throws Exception + */ + @Test + public void testSplitUnSplit() throws Exception { + splittedBytes.setChunksFromBytes(rawData); + + assertEquals(8, splittedBytes.getTotalSize()); + assertEquals(3, SplittedBytes.numberOfChunksNeededtoStore(rawData.length)); + + byte[] dataCopy = splittedBytes.getBytesFromChunks(); + + assertTrue("data after splitting and unplitting are not similar", Arrays.equals(rawData, dataCopy)); + + } + + /** + * This test shows that bytes can be split and + * unsplit by another instance of SplittedBytes + * @throws Exception + */ + @Test + public void testSplittedBytes2() throws Exception { + + SplittedBytes splittedBytes = new SplittedBytes(); + splittedBytes.setChunksFromBytes(rawData); + assertEquals(8, splittedBytes.getTotalSize()); + assertEquals(3, SplittedBytes.numberOfChunksNeededtoStore(rawData.length)); + + + + SplittedBytes otherSplittedBytes = new SplittedBytes(rawData.length); + for (BytesChunk bytesChunk : splittedBytes.getChunks()) { + assertFalse(otherSplittedBytes.isComplete()); + + int beforeAddActualSize = otherSplittedBytes.getActualSize(); + + otherSplittedBytes.addChunk(bytesChunk); + + int afterAddActualSize = otherSplittedBytes.getActualSize(); + + assertTrue("actualSize should progress", beforeAddActualSize < afterAddActualSize); + + } + + assertTrue(splittedBytes.isComplete()); + + byte[] dataCopy = splittedBytes.getBytesFromChunks(); + assertTrue("data after splitting and unplitting are not similar", Arrays.equals(rawData, dataCopy)); + + } + + /** + * This test shows that bytes can be split and + * unsplit by another instance of SplittedBytes + * @throws Exception + */ + @Test + public void testShuffleAdds() throws Exception { + + SplittedBytes splittedBytes = new SplittedBytes(); + splittedBytes.setChunksFromBytes(rawData); + + SplittedBytes otherSplittedBytes = new SplittedBytes(rawData.length); + + List<BytesChunk> someChunks = new ArrayList<BytesChunk>(); + + + // TODO use Collections.copy(someChunks, splittedBytes.getChunks()); + for (BytesChunk bytesChunk : splittedBytes.getChunks()) { + someChunks.add(bytesChunk); + } + + Collections.shuffle(someChunks); + + for (BytesChunk bytesChunk : someChunks) { + assertFalse(otherSplittedBytes.isComplete()); + otherSplittedBytes.addChunk(bytesChunk); + } + assertTrue(splittedBytes.isComplete()); + + byte[] dataCopy = splittedBytes.getBytesFromChunks(); + assertTrue("data after splitting and unplitting are not similar", Arrays.equals(rawData, dataCopy)); + + } + + + +}

1 0

r18 - trunk
by bleny＠users.nuiton.org May 4, 2010

May 4, 2010

Author: bleny Date: 2010-05-04 18:54:40 +0200 (Tue, 04 May 2010) New Revision: 18 Url: http://nuiton.org/repositories/revision/diswork/18 Log: module diswork-fs, recherche et transfert de fichiers Modified: trunk/pom.xml Modified: trunk/pom.xml =================================================================== --- trunk/pom.xml 2010-04-29 05:43:29 UTC (rev 17) +++ trunk/pom.xml 2010-05-04 16:54:40 UTC (rev 18) @@ -16,6 +16,10 @@ <artifactId>diswork</artifactId> <version>0.0.1-SNAPSHOT</version> + <modules> + <module>diswork-fs</module> + </modules> + <dependencies> <dependency> @@ -35,7 +39,7 @@ <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> - <version>4.7</version> + <version>4.8.1</version> <scope>test</scope> </dependency> @@ -109,7 +113,7 @@   - <packaging>jar</packaging> + <packaging>pom</packaging> <properties> <siteLocales>en</siteLocales>

1 0