|
|
@@ -0,0 +1,292 @@
|
|
|
+package cn.reghao.dfs.store.oss.service;
|
|
|
+
|
|
|
+import cn.reghao.dfs.store.meta.RocksClient;
|
|
|
+import cn.reghao.dfs.store.oss.db.mapper.DataBlockMapper;
|
|
|
+import cn.reghao.dfs.store.oss.db.mapper.FileMetaMapper;
|
|
|
+import cn.reghao.dfs.store.oss.model.ContentRange;
|
|
|
+import cn.reghao.dfs.store.oss.model.DataBlock;
|
|
|
+import cn.reghao.dfs.store.oss.model.FileMeta;
|
|
|
+import cn.reghao.dfs.store.service.FileStoreService;
|
|
|
+import cn.reghao.dfs.store.service.FileUrlService;
|
|
|
+import cn.reghao.jutil.jdk.security.DigestUtil;
|
|
|
+import cn.reghao.jutil.tool.id.IdGenerator;
|
|
|
+import cn.reghao.jutil.web.ServletUtil;
|
|
|
+import org.rocksdb.RocksDBException;
|
|
|
+import org.springframework.stereotype.Service;
|
|
|
+import org.springframework.util.StringUtils;
|
|
|
+import org.springframework.web.multipart.MultipartFile;
|
|
|
+
|
|
|
+import javax.servlet.http.HttpServletResponse;
|
|
|
+import java.io.*;
|
|
|
+import java.nio.charset.StandardCharsets;
|
|
|
+import java.util.*;
|
|
|
+
|
|
|
+/**
|
|
|
+ * @author reghao
|
|
|
+ * @date 2022-11-23 09:40:18
|
|
|
+ */
|
|
|
+@Service
|
|
|
+public class ObjectServiceImpl implements ObjectService {
|
|
|
+ private final RocksClient rocksClient;
|
|
|
+ private final IdGenerator objectIdGenerator;
|
|
|
+ private final IdGenerator blockIdGenerator;
|
|
|
+ private final FileMetaMapper fileMetaMapper;
|
|
|
+ private final DataBlockMapper dataBlockMapper;
|
|
|
+ // 20MiB
|
|
|
+ private final int blockSize = 1024*1024*20;
|
|
|
+ // 10MiB
|
|
|
+ private final long partLength = 1024*1024;
|
|
|
+ private FileStoreService fileStoreService;
|
|
|
+ private FileUrlService fileUrlService;
|
|
|
+
|
|
|
+ public ObjectServiceImpl(RocksClient rocksClient, FileMetaMapper fileMetaMapper, DataBlockMapper dataBlockMapper,
|
|
|
+ FileStoreService fileStoreService, FileUrlService fileUrlService) {
|
|
|
+ this.rocksClient = rocksClient;
|
|
|
+ this.objectIdGenerator = new IdGenerator(32, "object-id");
|
|
|
+ this.blockIdGenerator = new IdGenerator(32, "block-id");
|
|
|
+ this.fileMetaMapper = fileMetaMapper;
|
|
|
+ this.dataBlockMapper = dataBlockMapper;
|
|
|
+ this.fileStoreService = fileStoreService;
|
|
|
+ this.fileUrlService = fileUrlService;
|
|
|
+ }
|
|
|
+
|
|
|
+ public void putObject(String objectName, MultipartFile multipartFile) throws Exception {
|
|
|
+ String objectKey = String.format("/%s", objectName);
|
|
|
+ byte[] value = rocksClient.get(objectKey);
|
|
|
+ if (value != null) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ String contentType = multipartFile.getContentType();
|
|
|
+ long len = multipartFile.getSize();
|
|
|
+ InputStream inputStream = multipartFile.getInputStream();
|
|
|
+
|
|
|
+ String objectId = objectIdGenerator.stringId();
|
|
|
+
|
|
|
+ List<DataBlock> list = store(objectId, len, inputStream);
|
|
|
+ String sha256sum = DigestUtil.sha256sum(new FileInputStream(list.get(0).getAbsolutePath()));
|
|
|
+ FileMeta fileMeta = fileMetaMapper.findBySha256sum(sha256sum);
|
|
|
+ if (fileMeta == null) {
|
|
|
+ fileMeta = new FileMeta(objectName, objectId, len, contentType, sha256sum);
|
|
|
+ dataBlockMapper.saveAll(list);
|
|
|
+ fileMetaMapper.save(fileMeta);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private List<DataBlock> store(String objectId, long len, InputStream inputStream) throws IOException {
|
|
|
+ List<DataBlock> list = new ArrayList<>();
|
|
|
+ String absolutePath = fileUrlService.genFilePath(len, objectId, "dat");
|
|
|
+ fileStoreService.saveFile(absolutePath, inputStream);
|
|
|
+
|
|
|
+ String blockId = UUID.randomUUID().toString();
|
|
|
+ list.add(new DataBlock(objectId, 0, blockId, absolutePath));
|
|
|
+ return list;
|
|
|
+ }
|
|
|
+
|
|
|
+ private List<DataBlock> store1(String objectId, InputStream inputStream) throws IOException, RocksDBException {
|
|
|
+ List<DataBlock> list = new ArrayList<>();
|
|
|
+ byte[] buf = new byte[blockSize];
|
|
|
+ int index = 0;
|
|
|
+ int readLen;
|
|
|
+ while ((readLen = inputStream.read(buf, 0, blockSize)) != -1) {
|
|
|
+ //String blockId = blockIdGenerator.stringId();
|
|
|
+ String blockId = UUID.randomUUID().toString();
|
|
|
+ rocksClient.put(blockId, buf);
|
|
|
+ list.add(new DataBlock(objectId, index, blockId, ""));
|
|
|
+ index++;
|
|
|
+ }
|
|
|
+ return list;
|
|
|
+ }
|
|
|
+
|
|
|
+ public void postObject() {
|
|
|
+ }
|
|
|
+
|
|
|
+ public void initiateMultipartUpload() {
|
|
|
+ }
|
|
|
+
|
|
|
+ public void uploadPart() {
|
|
|
+ }
|
|
|
+
|
|
|
+ public void completeMultipartUpload() {
|
|
|
+ }
|
|
|
+
|
|
|
+ public void abortMultipartUpload() {
|
|
|
+ }
|
|
|
+
|
|
|
+ public void listParts() {
|
|
|
+ }
|
|
|
+
|
|
|
+ public void headObject(String objectName) throws RocksDBException {
|
|
|
+ }
|
|
|
+
|
|
|
+ public void getObject(String objectName) throws RocksDBException, IOException {
|
|
|
+ HttpServletResponse response = ServletUtil.getResponse();
|
|
|
+ FileMeta fileMeta = fileMetaMapper.findByObjectName(objectName);
|
|
|
+ if (fileMeta == null) {
|
|
|
+ response.setStatus(HttpServletResponse.SC_NOT_FOUND);
|
|
|
+ response.getOutputStream().write("".getBytes(StandardCharsets.UTF_8));
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ String objectId = fileMeta.getObjectId();
|
|
|
+
|
|
|
+ String contentType = fileMeta.getContentType();
|
|
|
+ long len = fileMeta.getSize();
|
|
|
+
|
|
|
+ String range = ServletUtil.getRequest().getHeader("range");
|
|
|
+ if (range != null) {
|
|
|
+ String rangeStr = StringUtils.trimAllWhitespace(range);
|
|
|
+ String[] arr = rangeStr.replace("bytes=", "").split("-");
|
|
|
+ long start = Long.parseLong(arr[0]);
|
|
|
+ if (arr.length == 1 && start == 0) {
|
|
|
+ writeAcceptRanges1(objectId, contentType, len);
|
|
|
+ } else {
|
|
|
+ ContentRange contentRange = parseContentRange(range, len);
|
|
|
+ byte[] bytes = range1(objectId, contentRange.getStart(), contentRange.getEnd());
|
|
|
+ writeContent(bytes, HttpServletResponse.SC_PARTIAL_CONTENT, contentType, contentRange, len);
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ writeAcceptRanges(objectId, contentType, len);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private void writeAcceptRanges(String objectId, String contentType, long len) throws IOException {
|
|
|
+ HttpServletResponse response = ServletUtil.getResponse();
|
|
|
+ response.setStatus(HttpServletResponse.SC_OK);
|
|
|
+ response.setContentType(contentType);
|
|
|
+ response.setHeader("Content-Length", ""+len);
|
|
|
+ response.setHeader("Accept-Ranges", "bytes");
|
|
|
+ //response.setHeader("Content-Range", "bytes "+0+"-"+(len-1)+"/"+len);
|
|
|
+
|
|
|
+ OutputStream outputStream = response.getOutputStream();
|
|
|
+ List<DataBlock> list = dataBlockMapper.findByObjectId(objectId);
|
|
|
+ for (DataBlock dataBlock : list) {
|
|
|
+ String blockId = dataBlock.getBlockId();
|
|
|
+ String absolutePath = dataBlock.getAbsolutePath();
|
|
|
+ FileInputStream fis = new FileInputStream(absolutePath);
|
|
|
+
|
|
|
+ // 1MiB
|
|
|
+ int bufSize = 1024*1024;
|
|
|
+ byte[] buf = new byte[bufSize];
|
|
|
+ int readLen;
|
|
|
+ while ((readLen = fis.read(buf, 0, bufSize)) != -1) {
|
|
|
+ outputStream.write(buf, 0, readLen);
|
|
|
+ outputStream.flush();
|
|
|
+ }
|
|
|
+ }
|
|
|
+ outputStream.close();
|
|
|
+ }
|
|
|
+
|
|
|
+ private void writeAcceptRanges1(String objectId, String contentType, long len) throws IOException {
|
|
|
+ HttpServletResponse response = ServletUtil.getResponse();
|
|
|
+ response.setStatus(HttpServletResponse.SC_OK);
|
|
|
+ response.setContentType(contentType);
|
|
|
+ response.setHeader("Content-Length", ""+partLength);
|
|
|
+ response.setHeader("Accept-Ranges", "bytes");
|
|
|
+ response.setHeader("Content-Range", "bytes "+0+"-"+(partLength)+"/"+len);
|
|
|
+
|
|
|
+ OutputStream outputStream = response.getOutputStream();
|
|
|
+ List<DataBlock> list = dataBlockMapper.findByObjectId(objectId);
|
|
|
+ for (DataBlock dataBlock : list) {
|
|
|
+ String blockId = dataBlock.getBlockId();
|
|
|
+ String absolutePath = dataBlock.getAbsolutePath();
|
|
|
+ FileInputStream fis = new FileInputStream(absolutePath);
|
|
|
+
|
|
|
+ // 1MiB
|
|
|
+ int bufSize = 1024*1024;
|
|
|
+ byte[] buf = new byte[bufSize];
|
|
|
+ int readLen;
|
|
|
+ while ((readLen = fis.read(buf, 0, bufSize)) != -1) {
|
|
|
+ outputStream.write(buf, 0, readLen);
|
|
|
+ outputStream.flush();
|
|
|
+ }
|
|
|
+ }
|
|
|
+ outputStream.close();
|
|
|
+ }
|
|
|
+
|
|
|
+ private ContentRange parseContentRange(String range, long len) {
|
|
|
+ String rangeStr = StringUtils.trimAllWhitespace(range);
|
|
|
+ String[] arr = rangeStr.replace("bytes=", "").split("-");
|
|
|
+ long start = Long.parseLong(arr[0]);
|
|
|
+ long end = 0;
|
|
|
+ if (arr.length == 2) {
|
|
|
+ end = Long.parseLong(arr[1]);
|
|
|
+ } else {
|
|
|
+ long l = len - start;
|
|
|
+ if (l > 0) {
|
|
|
+ end = Math.min(l, partLength) + start;
|
|
|
+ } else {
|
|
|
+ System.out.println();
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ return new ContentRange(start, end);
|
|
|
+ }
|
|
|
+
|
|
|
+ private void writeContent(byte[] bytes, int statusCode, String contentType, ContentRange contentRange, long len) throws IOException {
|
|
|
+ HttpServletResponse response = ServletUtil.getResponse();
|
|
|
+ response.setStatus(statusCode);
|
|
|
+ response.setContentType(contentType);
|
|
|
+ response.setHeader("Content-Length", ""+(contentRange.getEnd()-contentRange.getStart()));
|
|
|
+ response.setHeader("Accept-Ranges", "bytes");
|
|
|
+ response.setHeader("Content-Range", "bytes "+contentRange.getStart()+"-"+contentRange.getEnd()+"/"+len);
|
|
|
+
|
|
|
+ OutputStream outputStream = response.getOutputStream();
|
|
|
+ outputStream.write(bytes);
|
|
|
+ outputStream.flush();
|
|
|
+ outputStream.close();
|
|
|
+ }
|
|
|
+
|
|
|
+ private byte[] range1(String objectId, long start, long end) throws IOException {
|
|
|
+ String absolutePath = dataBlockMapper.findByObjectId(objectId).get(0).getAbsolutePath();
|
|
|
+ RandomAccessFile raf = new RandomAccessFile(absolutePath, "r");
|
|
|
+ raf.seek(start);
|
|
|
+ long bufSize = end - start;
|
|
|
+ byte[] bytes = new byte[(int) bufSize];
|
|
|
+ int readBytes = raf.read(bytes);
|
|
|
+ return bytes;
|
|
|
+ }
|
|
|
+
|
|
|
+ private byte[] range(String objectId, long start, long end) throws RocksDBException {
|
|
|
+ /*int index1 = (int)(start/blockSize + start%blockSize);
|
|
|
+ int index2 = (int)(end/blockSize + end%blockSize);
|
|
|
+
|
|
|
+ Map<Integer, DataBlock> map = dataBlockMapper.findByObjectId(objectId).stream()
|
|
|
+ .collect(Collectors.toMap(DataBlock::getIndex, Function.identity()));
|
|
|
+ byte[] result = new byte[0];
|
|
|
+ for (int i = index1; i < index2; i++) {
|
|
|
+ DataBlock dataBlock = map.get(index1);
|
|
|
+ String blockId = dataBlock.getBlockId();
|
|
|
+ byte[] bytes = rocksClient.get(blockId);
|
|
|
+ result = mergeByteArray(result, bytes);
|
|
|
+ }*/
|
|
|
+
|
|
|
+ List<DataBlock> list = dataBlockMapper.findByObjectId(objectId);
|
|
|
+ byte[] result = new byte[0];
|
|
|
+ for (int i = 0; i < 10; i++) {
|
|
|
+ DataBlock dataBlock = list.get(i);
|
|
|
+ String blockId = dataBlock.getBlockId();
|
|
|
+ byte[] bytes = rocksClient.get(blockId);
|
|
|
+ result = mergeByteArray(result, bytes);
|
|
|
+ }
|
|
|
+ /*for (DataBlock dataBlock : list) {
|
|
|
+ String blockId = dataBlock.getBlockId();
|
|
|
+ byte[] bytes = rocksClient.get(blockId);
|
|
|
+ result = mergeByteArray(result, bytes);
|
|
|
+ }*/
|
|
|
+
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ private byte[] mergeByteArray(byte[] arr1, byte[] arr2) {
|
|
|
+ byte[] arr3 = new byte[arr1.length + arr2.length];
|
|
|
+ System.arraycopy(arr1, 0, arr3, 0, arr1.length);
|
|
|
+ System.arraycopy(arr2, 0, arr3, arr1.length, arr2.length);
|
|
|
+ return arr3;
|
|
|
+ }
|
|
|
+
|
|
|
+ public void deleteObject() {
|
|
|
+ }
|
|
|
+
|
|
|
+ public void deleteMultipleObjects() {
|
|
|
+ }
|
|
|
+}
|