-
Notifications
You must be signed in to change notification settings - Fork 34
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(s3stream): add a third party package
moe.cnkirito.kdio (#654)
Signed-off-by: Ning Yu <[email protected]>
- Loading branch information
1 parent
8f09da5
commit ee9e776
Showing
9 changed files
with
819 additions
and
34 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
76 changes: 76 additions & 0 deletions
76
s3stream/src/main/java/com/automq/stream/thirdparty/moe/cnkirito/kdio/DirectChannel.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,76 @@ | ||
| /** | ||
| * Copyright 2019 xujingfeng ([email protected]) | ||
| * <p> | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * <p> | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * <p> | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package com.automq.stream.thirdparty.moe.cnkirito.kdio; | ||
|
|
||
| import java.io.IOException; | ||
| import java.nio.ByteBuffer; | ||
| import java.nio.channels.Channel; | ||
|
|
||
| public interface DirectChannel extends Channel { | ||
| /** | ||
| * Writes from the <tt>src</tt> buffer into this channel at <tt>position</tt>. | ||
| * | ||
| * @param src | ||
| * The {@link ByteBuffer} to write from | ||
| * | ||
| * @param position | ||
| * The position within the file at which to start writing | ||
| * | ||
| * @return How many bytes were written from <tt>src</tt> into the file | ||
| * @throws IOException | ||
| */ | ||
| int write(ByteBuffer src, long position) throws IOException; | ||
|
|
||
| /** | ||
| * Reads from this channel into the <tt>dst</tt> buffer from <tt>position</tt>. | ||
| * | ||
| * @param dst | ||
| * The {@link ByteBuffer} to read into | ||
| * | ||
| * @param position | ||
| * The position within the file at which to start reading | ||
| * | ||
| * @return How many bytes were placed into <tt>dst</tt> | ||
| * @throws IOException | ||
| */ | ||
| int read(ByteBuffer dst, long position) throws IOException; | ||
|
|
||
| /** | ||
| * @return The file size for this channel | ||
| */ | ||
| long size(); | ||
|
|
||
| /** | ||
| * @return <tt>true</tt> if this channel is read only, <tt>false</tt> otherwise | ||
| */ | ||
| boolean isReadOnly(); | ||
|
|
||
| /** | ||
| * Truncates this file's length to <tt>fileLength</tt>. | ||
| * | ||
| * @param fileLength The length to which to truncate | ||
| * | ||
| * @return This UnsafeByteAlignedChannel | ||
| * | ||
| * @throws IOException | ||
| */ | ||
| DirectChannel truncate(long fileLength) throws IOException; | ||
|
|
||
| /** | ||
| * @return The file descriptor for this channel | ||
| */ | ||
| int getFD(); | ||
| } |
128 changes: 128 additions & 0 deletions
128
s3stream/src/main/java/com/automq/stream/thirdparty/moe/cnkirito/kdio/DirectChannelImpl.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,128 @@ | ||
| /** | ||
| * Copyright 2019 xujingfeng ([email protected]) | ||
| * <p> | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * <p> | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * <p> | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package com.automq.stream.thirdparty.moe.cnkirito.kdio; | ||
|
|
||
| import java.io.File; | ||
| import java.io.IOException; | ||
| import java.nio.ByteBuffer; | ||
| import java.nio.channels.ClosedChannelException; | ||
| import java.nio.channels.NonWritableChannelException; | ||
|
|
||
| public class DirectChannelImpl implements DirectChannel { | ||
| private final DirectIOLib lib; | ||
| private final int fd; | ||
| private boolean isOpen; | ||
| private long fileLength; | ||
| private final boolean isReadOnly; | ||
|
|
||
| private DirectChannelImpl(DirectIOLib lib, int fd, long fileLength, boolean readOnly) { | ||
| this.lib = lib; | ||
| this.fd = fd; | ||
| this.isOpen = true; | ||
| this.isReadOnly = readOnly; | ||
| this.fileLength = fileLength; | ||
| } | ||
|
|
||
| public static DirectChannel getChannel(File file, boolean readOnly) throws IOException { | ||
| DirectIOLib lib = DirectIOLib.getLibForPath(file.toString()); | ||
| if (null == lib) { | ||
| throw new IOException("No DirectIOLib found for path " + file); | ||
| } | ||
| return getChannel(lib, file, readOnly); | ||
| } | ||
|
|
||
| public static DirectChannel getChannel(DirectIOLib lib, File file, boolean readOnly) throws IOException { | ||
| int fd = lib.oDirectOpen(file.toString(), readOnly); | ||
| long length = file.length(); | ||
| return new DirectChannelImpl(lib, fd, length, readOnly); | ||
| } | ||
|
|
||
| private void ensureOpen() throws ClosedChannelException { | ||
| if (!isOpen()) { | ||
| throw new ClosedChannelException(); | ||
| } | ||
| } | ||
|
|
||
| private void ensureWritable() { | ||
| if (isReadOnly()) { | ||
| throw new NonWritableChannelException(); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public int read(ByteBuffer dst, long position) throws IOException { | ||
| ensureOpen(); | ||
| return lib.pread(fd, dst, position); | ||
| } | ||
|
|
||
| @Override | ||
| public int write(ByteBuffer src, long position) throws IOException { | ||
| ensureOpen(); | ||
| ensureWritable(); | ||
| assert src.position() == lib.blockStart(src.position()); | ||
|
|
||
| int written = lib.pwrite(fd, src, position); | ||
|
|
||
| // update file length if we wrote past it | ||
| fileLength = Math.max(position + written, fileLength); | ||
| return written; | ||
| } | ||
|
|
||
| @Override | ||
| public DirectChannel truncate(final long length) throws IOException { | ||
| ensureOpen(); | ||
| ensureWritable(); | ||
| if (DirectIOLib.ftruncate(fd, length) < 0) { | ||
| throw new IOException("Error during truncate on descriptor " + fd + ": " + | ||
| DirectIOLib.getLastError()); | ||
| } | ||
| fileLength = length; | ||
| return this; | ||
| } | ||
|
|
||
| @Override | ||
| public long size() { | ||
| return fileLength; | ||
| } | ||
|
|
||
| @Override | ||
| public int getFD() { | ||
| return fd; | ||
| } | ||
|
|
||
|
|
||
| @Override | ||
| public boolean isOpen() { | ||
| return isOpen; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isReadOnly() { | ||
| return isReadOnly; | ||
| } | ||
|
|
||
| @Override | ||
| public void close() throws IOException { | ||
| if (!isOpen()) { | ||
| return; | ||
| } | ||
| isOpen = false; | ||
| if (lib.close(fd) < 0) { | ||
| throw new IOException("Error closing file with descriptor " + fd + ": " + | ||
| DirectIOLib.getLastError()); | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.