Skip to content

Commit

Permalink
Java: SeaweedOutputStream add option to pass in collection
Browse files Browse the repository at this point in the history
  • Loading branch information
chrislusf committed Aug 17, 2023
1 parent 2af4cab commit 4523cd7
Show file tree
Hide file tree
Showing 2 changed files with 13 additions and 7 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ public class SeaweedOutputStream extends OutputStream {
private ByteBuffer buffer;
private long outputIndex;
private String replication = "";
private String collection = "";

public SeaweedOutputStream(FilerClient filerClient, final String fullpath) {
this(filerClient, fullpath, "");
Expand All @@ -53,7 +54,6 @@ public SeaweedOutputStream(FilerClient filerClient, final String path, FilerProt
this.lastFlushOffset = 0;
this.bufferSize = bufferSize;
this.buffer = ByteBufferPool.request(bufferSize);
this.outputIndex = 0;
this.writeOperations = new ConcurrentLinkedDeque<>();

this.maxConcurrentRequestCount = Runtime.getRuntime().availableProcessors();
Expand Down Expand Up @@ -83,6 +83,13 @@ public SeaweedOutputStream(FilerClient filerClient, final String path, FilerProt

}

public void setReplication(String replication) {
this.replication = replication;
}
public void setCollection(String collection) {
this.collection = collection;
}

public static String getParentDirectory(String path) {
int protoIndex = path.indexOf("://");
if (protoIndex >= 0) {
Expand Down Expand Up @@ -144,13 +151,11 @@ public synchronized void write(final byte[] data, final int off, final int lengt

if (numberOfBytesToWrite < writableBytes) {
buffer.put(data, currentOffset, numberOfBytesToWrite);
outputIndex += numberOfBytesToWrite;
break;
}

// System.out.println(path + " [" + (outputIndex + currentOffset) + "," + ((outputIndex + currentOffset) + writableBytes) + ") " + buffer.capacity());
buffer.put(data, currentOffset, writableBytes);
outputIndex += writableBytes;
currentOffset += writableBytes;
writeCurrentBufferToService();
numberOfBytesToWrite = numberOfBytesToWrite - writableBytes;
Expand Down Expand Up @@ -194,7 +199,6 @@ public synchronized void close() throws IOException {
lastError = new IOException("Stream is closed!");
ByteBufferPool.release(buffer);
buffer = null;
outputIndex = 0;
closed = true;
writeOperations.clear();
if (!threadExecutor.isShutdown()) {
Expand Down Expand Up @@ -225,7 +229,7 @@ private synchronized int submitWriteBufferToService(final ByteBuffer bufferToWri
}
final Future<Void> job = completionService.submit(() -> {
// System.out.println(path + " is going to save [" + (writePosition) + "," + ((writePosition) + bytesLength) + ")");
SeaweedWrite.writeData(entry, replication, filerClient, writePosition, bufferToWrite.array(), bufferToWrite.position(), bufferToWrite.limit(), path);
SeaweedWrite.writeData(entry, replication, collection, filerClient, writePosition, bufferToWrite.array(), bufferToWrite.position(), bufferToWrite.limit(), path);
// System.out.println(path + " saved [" + (writePosition) + "," + ((writePosition) + bytesLength) + ")");
ByteBufferPool.release(bufferToWrite);
return null;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ public class SeaweedWrite {

public static void writeData(FilerProto.Entry.Builder entry,
final String replication,
String collection,
final FilerClient filerClient,
final long offset,
final byte[] bytes,
Expand All @@ -36,7 +37,7 @@ public static void writeData(FilerProto.Entry.Builder entry,
for (long waitTime = 1000L; waitTime < 10 * 1000; waitTime += waitTime / 2) {
try {
FilerProto.FileChunk.Builder chunkBuilder = writeChunk(
replication, filerClient, offset, bytes, bytesOffset, bytesLength, path);
replication, collection, filerClient, offset, bytes, bytesOffset, bytesLength, path);
lastException = null;
synchronized (entry) {
entry.addChunks(chunkBuilder);
Expand All @@ -59,6 +60,7 @@ public static void writeData(FilerProto.Entry.Builder entry,
}

public static FilerProto.FileChunk.Builder writeChunk(final String replication,
final String collection,
final FilerClient filerClient,
final long offset,
final byte[] bytes,
Expand All @@ -67,7 +69,7 @@ public static FilerProto.FileChunk.Builder writeChunk(final String replication,
final String path) throws IOException {
FilerProto.AssignVolumeResponse response = filerClient.getBlockingStub().assignVolume(
FilerProto.AssignVolumeRequest.newBuilder()
.setCollection(filerClient.getCollection())
.setCollection(Strings.isNullOrEmpty(collection) ? filerClient.getCollection() : collection)
.setReplication(Strings.isNullOrEmpty(replication) ? filerClient.getReplication() : replication)
.setDataCenter("")
.setTtlSec(0)
Expand Down

0 comments on commit 4523cd7

Please sign in to comment.