1. package com.imooc.bigdata.hos.server;
2.
3. import java.io.ByteArrayInputStream;
4. import java.io.IOException;
5. import java.io.InputStream;
6. import java.nio.ByteBuffer;
7. import java.util.ArrayList;
8. import java.util.Collections;
9. import java.util.HashMap;
10. import java.util.List;
11. import java.util.Map;
12.
13. import org.apache.curator.framework.CuratorFramework;
14. import org.apache.curator.framework.CuratorFrameworkFactory;
15. import org.apache.curator.framework.recipes.locks.InterProcessMutex;
16. import org.apache.curator.retry.ExponentialBackoffRetry;
17. import org.apache.hadoop.hbase.Cell;
18. import org.apache.hadoop.hbase.CellUtil;
19. import org.apache.hadoop.hbase.TableName;
20. import org.apache.hadoop.hbase.client.Admin;
21. import org.apache.hadoop.hbase.client.Connection;
22. import org.apache.hadoop.hbase.client.Get;
23. import org.apache.hadoop.hbase.client.Put;
24. import org.apache.hadoop.hbase.client.Result;
25. import org.apache.hadoop.hbase.client.ResultScanner;
26. import org.apache.hadoop.hbase.client.Scan;
27. import org.apache.hadoop.hbase.filter.BinaryComparator;
28. import org.apache.hadoop.hbase.filter.ColumnPrefixFilter;
29. import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
30. import org.apache.hadoop.hbase.filter.FilterList;
31. import org.apache.hadoop.hbase.filter.FilterList.Operator;
32. import org.apache.hadoop.hbase.filter.PageFilter;
33. import org.apache.hadoop.hbase.filter.QualifierFilter;
34. import org.apache.hadoop.hbase.io.ByteBufferInputStream;
35. import org.apache.hadoop.hbase.util.Bytes;
36. import org.apache.log4j.Logger;
37.
38. import com.google.common.base.Strings;
39. import com.imooc.bigdata.hos.common.HosObject;
40. import com.imooc.bigdata.hos.common.HosObjectSummary;
41. import com.imooc.bigdata.hos.common.ObjectListResult;
42. import com.imooc.bigdata.hos.common.ObjectMetaData;
43. import com.imooc.bigdata.hos.common.util.JsonUtil;
44.
45. /**
46. * Created by jixin on 18-3-13.
47. */
48. public class HosStoreServiceImpl implements IHosStoreService {
49.
50. private static Logger logger = Logger.getLogger(HosStoreServiceImpl.class);
51. private Connection connection = null;
52. private IHdfsService fileStore;
53. private String zkUrls;
54. private CuratorFramework zkClient;
55.
56. public HosStoreServiceImpl(Connection connection, IHdfsService fileStore, String zkurls)
57. throws IOException {
58. this.connection = connection;
59. this.fileStore = fileStore;
60. this.zkUrls = zkurls;
61. zkClient = CuratorFrameworkFactory.newClient(zkUrls, new ExponentialBackoffRetry(20, 5));
62. zkClient.start();
63. }
64.
65. @Override
66. public void createSeqTable() throws IOException {
67. Admin admin = connection.getAdmin();
68. TableName tableName = TableName.valueOf(HosUtil.BUCKET_DIR_SEQ_TABLE);
69. if (admin.tableExists(tableName)) {
70. return;
71. }
72. HBaseService.createTable(connection, HosUtil.BUCKET_DIR_SEQ_TABLE,
73. new String[]{HosUtil.BUCKET_DIR_SEQ_CF});
74. }
75.
76. @Override
77. public void put(String bucket, String key, ByteBuffer input, long length, String mediaType,
78. Map<String, String> properties) throws Exception {
79. InterProcessMutex lock = null;
80. try {
81. if (key.endsWith("/")) {
82. //put dir object
83. putDir(bucket, key);
84. return;
85. }
86. //put dir
87. String dir = key.substring(0, key.lastIndexOf("/") + 1);
88. String hash = null;
89. while (hash == null) {
90. if (!dirExist(bucket, dir)) {
91. hash = putDir(bucket, dir);
92. } else {
93. hash = this.getDirSeqId(bucket, dir);
94. }
95. }
96. String lockey = key.replaceAll("/", "_");
97. lock = new InterProcessMutex(this.zkClient, "/mos/" + bucket + "/" + lockey);
98. lock.acquire();
99. //put
100. String fileKey = hash + "_" + key.substring(key.lastIndexOf("/") + 1);
101. Put contentPut = new Put(fileKey.getBytes());
102. if (!Strings.isNullOrEmpty(mediaType)) {
103. contentPut.addColumn(HosUtil.OBJ_META_CF_BYTES,
104. HosUtil.OBJ_MEDIATYPE_QUALIFIER, mediaType.getBytes());
105. }
106. if (properties != null) {
107. String props = JsonUtil.toJson(properties);
108. contentPut.addColumn(HosUtil.OBJ_META_CF_BYTES,
109. HosUtil.OBJ_PROPS_QUALIFIER, props.getBytes());
110. }
111. contentPut.addColumn(HosUtil.OBJ_META_CF_BYTES,
112. HosUtil.OBJ_LEN_QUALIFIER, Bytes.toBytes((long) length));
113.
114. if (length <= HosUtil.FILE_STORE_THRESHOLD) {
115. ByteBuffer qualifierBuffer = ByteBuffer.wrap(HosUtil.OBJ_CONT_QUALIFIER);
116. contentPut.addColumn(HosUtil.OBJ_CONT_CF_BYTES,
117. qualifierBuffer, System.currentTimeMillis(), input);
118. qualifierBuffer.clear();
119. }
120. HBaseService.putRow(connection, HosUtil.getObjTableName(bucket), contentPut);
121. if (length > HosUtil.FILE_STORE_THRESHOLD) {
122. String fileDir = HosUtil.FILE_STORE_ROOT + "/" + bucket + "/" + hash;
123. String name = key.substring(key.lastIndexOf("/") + 1);
124. InputStream inputStream = new ByteBufferInputStream(input);
125. this.fileStore.saveFile(fileDir, name, inputStream, length, getBucketReplication(bucket));
126. }
127. } finally {
128. if (lock != null) {
129. lock.release();
130. }
131. }
132. }
133.
134. @Override
135. public HosObjectSummary getSummary(String bucket, String key) throws IOException {
136. if (key.endsWith("/")) {
137. Result result = HBaseService
138. .getRow(connection, HosUtil.getDirTableName(bucket), key);
139. if (!result.isEmpty()) {
140. return this.dirObjectToSummary(result, bucket, key);
141. } else {
142. return null;
143. }
144. }
145. String dir = key.substring(0, key.lastIndexOf("/") + 1);
146. String seq = this.getDirSeqId(bucket, dir);
147. if (seq == null) {
148. return null;
149. }
150. //序列号_文件名
151. String objKey = seq + "_" + key.substring(key.lastIndexOf("/") + 1);
152. Result result = HBaseService
153. .getRow(connection, HosUtil.getObjTableName(bucket), objKey);
154. if (result.isEmpty()) {
155. return null;
156. }
157. return this.resultToObjectSummary(result, bucket, dir);
158. }
159.
160. @Override
161. public List<HosObjectSummary> list(String bucket, String startKey, String endKey)
162. throws IOException {
163.
164. String dir1 = startKey.substring(0, startKey.lastIndexOf("/") + 1).trim();
165. if (dir1.length() == 0) {
166. dir1 = "/";
167. }
168. String dir2 = endKey.substring(0, startKey.lastIndexOf("/") + 1).trim();
169. if (dir2.length() == 0) {
170. dir2 = "/";
171. }
172. String name1 = startKey.substring(startKey.lastIndexOf("/") + 1);
173. String name2 = endKey.substring(startKey.lastIndexOf("/") + 1);
174. String seqId = this.getDirSeqId(bucket, dir1);
175. //查询dir1中大于name1的全部文件
176. List<HosObjectSummary> keys = new ArrayList<>();
177. if (seqId != null && name1.length() > 0) {
178. byte[] max = Bytes.createMaxByteArray(100);
179. byte[] tail = Bytes.add(Bytes.toBytes(seqId), max);
180. if (dir1.equals(dir2)) {
181. tail = (seqId + "_" + name2).getBytes();
182. }
183. byte[] start = (seqId + "_" + name1).getBytes();
184. ResultScanner scanner1 = HBaseService
185. .scanner(connection, HosUtil.getObjTableName(bucket), start, tail);
186. Result result = null;
187. while ((result = scanner1.next()) != null) {
188. HosObjectSummary summary = this.resultToObjectSummary(result, bucket, dir1);
189. keys.add(summary);
190. }
191. if (scanner1 != null) {
192. scanner1.close();
193. }
194. }
195. //startkey~endkey之间的全部目录
196. ResultScanner scanner2 = HBaseService
197. .scanner(connection, HosUtil.getDirTableName(bucket), startKey, endKey);
198. Result result = null;
199. while ((result = scanner2.next()) != null) {
200. String seqId2 = Bytes.toString(result.getValue(HosUtil.DIR_META_CF_BYTES,
201. HosUtil.DIR_SEQID_QUALIFIER));
202. if (seqId2 == null) {
203. continue;
204. }
205. String dir = Bytes.toString(result.getRow());
206. keys.add(dirObjectToSummary(result, bucket, dir));
207. getDirAllFiles(bucket, dir, seqId2, keys, endKey);
208. }
209. if (scanner2 != null) {
210. scanner2.close();
211. }
212. Collections.sort(keys);
213. return keys;
214. }
215.
216. @Override
217. public ObjectListResult listDir(String bucket, String dir, String start, int maxCount)
218. throws IOException {
219. if (start == null) {
220. start = "";
221. }
222. //查询目录表
223. Get get = new Get(Bytes.toBytes(dir));
224. get.addFamily(HosUtil.DIR_SUBDIR_CF_BYTES);
225. if (start.length() > 0) {
226. get.setFilter(new QualifierFilter(CompareOp.GREATER_OR_EQUAL,
227. new BinaryComparator(Bytes.toBytes(start))));
228. }
229. int maxCount1 = maxCount + 2;
230. Result dirResult = HBaseService
231. .getRow(connection, HosUtil.getDirTableName(bucket), get);
232. List<HosObjectSummary> subDirs = null;
233. if (!dirResult.isEmpty()) {
234. subDirs = new ArrayList<>();
235. for (Cell cell : dirResult.rawCells()) {
236. HosObjectSummary summary = new HosObjectSummary();
237. byte[] qualifierBytes = new byte[cell.getQualifierLength()];
238. CellUtil.copyQualifierTo(cell, qualifierBytes, 0);
239. String name = Bytes.toString(qualifierBytes);
240. summary.setKey(dir + name + "/");
241. summary.setName(name);
242. summary.setLastModifyTime(cell.getTimestamp());
243. summary.setMediaType("");
244. summary.setBucket(bucket);
245. summary.setLength(0);
246. subDirs.add(summary);
247. if (subDirs.size() >= maxCount1) {
248. break;
249. }
250. }
251. }
252.
253. //查询文件表
254. String dirSeq = this.getDirSeqId(bucket, dir);
255. byte[] objStart = Bytes.toBytes(dirSeq + "_" + start);
256. Scan objScan = new Scan();
257. objScan.setRowPrefixFilter(Bytes.toBytes(dirSeq + "_"));
258. /**
259. * 有这样一个场景，在HBase中需要分页查询，同时根据某一列的值进行过滤。
260. * 不同于RDBMS天然支持分页查询，HBase要进行分页必须由自己实现。据我了解的，目前有两种方案，
261. * 一是《HBase权威指南》中提到的用PageFilter加循环动态设置startRow实现，
262. * 详细见这里。但这种方法效率比较低，且有冗余查询。
263. * 因此京东研发了一种用额外的一张表来保存行序号的方案。 该种方案效率较高，但实现麻烦些，需要维护一张额外的表
264. * */
265. objScan.setFilter(new PageFilter(maxCount + 1));
266. objScan.setStartRow(objStart);
267. objScan.setMaxResultsPerColumnFamily(maxCount1);
268. objScan.addFamily(HosUtil.OBJ_META_CF_BYTES);
269. logger.info("scan start: " + Bytes.toString(objStart) + " - ");
270. ResultScanner objScanner = HBaseService
271. .scanner(connection, HosUtil.getObjTableName(bucket), objScan);
272. List<HosObjectSummary> objectSummaryList = new ArrayList<>();
273. Result result = null;
274. while (objectSummaryList.size() < maxCount1 && (result = objScanner.next()) != null) {
275. HosObjectSummary summary = this.resultToObjectSummary(result, bucket, dir);
276. objectSummaryList.add(summary);
277. }
278. if (objScanner != null) {
279. objScanner.close();
280. }
281.
282.
283. //返回给用户maxCount
284. logger.info("scan complete: " + Bytes.toString(objStart) + " - ");
285. if (subDirs != null && subDirs.size() > 0) {
286. objectSummaryList.addAll(subDirs);
287. }
288. Collections.sort(objectSummaryList);
289. ObjectListResult listResult = new ObjectListResult();
290. HosObjectSummary nextMarkerObj =
291. objectSummaryList.size() > maxCount ? objectSummaryList.get(objectSummaryList.size() - 1)
292. : null;
293. if (nextMarkerObj != null) {
294. listResult.setNextMarker(nextMarkerObj.getKey());
295. }
296. if (objectSummaryList.size() > maxCount) {
297. objectSummaryList = objectSummaryList.subList(0, maxCount);
298. }
299. listResult.setMaxKeyNumber(maxCount);
300. if (objectSummaryList.size() > 0) {
301. listResult.setMinKey(objectSummaryList.get(0).getKey());
302. listResult.setMaxKey(objectSummaryList.get(objectSummaryList.size() - 1).getKey());
303. }
304. listResult.setObjectCount(objectSummaryList.size());
305. listResult.setObjectList(objectSummaryList);
306. listResult.setBucket(bucket);
307.
308. return listResult;
309. }
310.
311. @Override
312. public ObjectListResult listByPrefix(String bucket, String dir, String keyPrefix, String start,
313. int maxCount) throws IOException {
314. if (start == null) {
315. start = "";
316. }
317. FilterList filterList = new FilterList(Operator.MUST_PASS_ALL);
318. filterList.addFilter(new ColumnPrefixFilter(keyPrefix.getBytes()));
319. if (start.length() > 0) {
320. filterList.addFilter(new QualifierFilter(CompareOp.GREATER_OR_EQUAL,
321. new BinaryComparator(Bytes.toBytes(start))));
322. }
323. int maxCount1 = maxCount + 2;
324. Result dirResult = HBaseService
325. .getRow(connection, HosUtil.getDirTableName(bucket), dir, filterList);
326. List<HosObjectSummary> subDirs = null;
327. if (!dirResult.isEmpty()) {
328. subDirs = new ArrayList<>();
329. for (Cell cell : dirResult.rawCells()) {
330. HosObjectSummary summary = new HosObjectSummary();
331. byte[] qualifierBytes = new byte[cell.getQualifierLength()];
332. CellUtil.copyQualifierTo(cell, qualifierBytes, 0);
333. String name = Bytes.toString(qualifierBytes);
334. summary.setKey(dir + name + "/");
335. summary.setName(name);
336. summary.setLastModifyTime(cell.getTimestamp());
337. summary.setMediaType("");
338. summary.setBucket(bucket);
339. summary.setLength(0);
340. subDirs.add(summary);
341. if (subDirs.size() >= maxCount1) {
342. break;
343. }
344. }
345. }
346.
347. String dirSeq = this.getDirSeqId(bucket, dir);
348. byte[] objStart = Bytes.toBytes(dirSeq + "_" + start);
349. Scan objScan = new Scan();
350. objScan.setRowPrefixFilter(Bytes.toBytes(dirSeq + "_" + keyPrefix));
351. objScan.setFilter(new PageFilter(maxCount + 1));
352. objScan.setStartRow(objStart);
353. objScan.setMaxResultsPerColumnFamily(maxCount1);
354. objScan.addFamily(HosUtil.OBJ_META_CF_BYTES);
355. logger.info("scan start: " + Bytes.toString(objStart) + " - ");
356. ResultScanner objScanner = HBaseService
357. .scanner(connection, HosUtil.getObjTableName(bucket), objScan);
358. List<HosObjectSummary> objectSummaryList = new ArrayList<>();
359. Result result = null;
360. while (objectSummaryList.size() < maxCount1 && (result = objScanner.next()) != null) {
361. HosObjectSummary summary = this.resultToObjectSummary(result, bucket, dir);
362. objectSummaryList.add(summary);
363. }
364. if (objScanner != null) {
365. objScanner.close();
366. }
367. logger.info("scan complete: " + Bytes.toString(objStart) + " - ");
368. if (subDirs != null && subDirs.size() > 0) {
369. objectSummaryList.addAll(subDirs);
370. }
371. Collections.sort(objectSummaryList);
372. ObjectListResult listResult = new ObjectListResult();
373. HosObjectSummary nextMarkerObj =
374. objectSummaryList.size() > maxCount ? objectSummaryList.get(objectSummaryList.size() - 1)
375. : null;
376. if (nextMarkerObj != null) {
377. listResult.setNextMarker(nextMarkerObj.getKey());
378. }
379. if (objectSummaryList.size() > maxCount) {
380. objectSummaryList = objectSummaryList.subList(0, maxCount);
381. }
382. listResult.setMaxKeyNumber(maxCount);
383. if (objectSummaryList.size() > 0) {
384. listResult.setMinKey(objectSummaryList.get(0).getKey());
385. listResult.setMaxKey(objectSummaryList.get(objectSummaryList.size() - 1).getKey());
386. }
387. listResult.setObjectCount(objectSummaryList.size());
388. listResult.setObjectList(objectSummaryList);
389. listResult.setBucket(bucket);
390.
391. return listResult;
392. }
393.
394. @Override
395. public HosObject getObject(String bucket, String key) throws IOException {
396. if (key.endsWith("/")) {
397. Result result = HBaseService
398. .getRow(connection, HosUtil.getDirTableName(bucket), key);
399. if (result.isEmpty()) {
400. return null;
401. }
402. ObjectMetaData metaData = new ObjectMetaData();
403. metaData.setBucket(bucket);
404. metaData.setKey(key);
405. metaData.setLastModifyTime(result.rawCells()[0].getTimestamp());
406. metaData.setLength(0);
407. HosObject object = new HosObject();
408. object.setMetaData(metaData);
409. return object;
410. }
411. String dir = key.substring(0, key.lastIndexOf("/") + 1);
412. String name = key.substring(key.lastIndexOf("/") + 1);
413. String seq = this.getDirSeqId(bucket, dir);
414. String objKey = seq + "_" + name;
415. Result result = HBaseService
416. .getRow(connection, HosUtil.getObjTableName(bucket), objKey);
417. if (result.isEmpty()) {
418. return null;
419. }
420. HosObject object = new HosObject();
421. if (result.containsNonEmptyColumn(HosUtil.OBJ_CONT_CF_BYTES,
422. HosUtil.OBJ_CONT_QUALIFIER)) {
423. ByteArrayInputStream bas = new ByteArrayInputStream(
424. result
425. .getValue(HosUtil.OBJ_CONT_CF_BYTES,
426. HosUtil.OBJ_CONT_QUALIFIER));
427. object.setContent(bas);
428. } else {
429. String fileDir = HosUtil.FILE_STORE_ROOT + "/" + bucket + "/" + seq;
430. InputStream inputStream = this.fileStore.openFile(fileDir, name);
431. object.setContent(inputStream);
432. }
433. long len = Bytes.toLong(result.getValue(HosUtil.OBJ_META_CF_BYTES,
434. HosUtil.OBJ_LEN_QUALIFIER));
435. ObjectMetaData metaData = new ObjectMetaData();
436. metaData.setBucket(bucket);
437. metaData.setKey(key);
438. metaData.setLastModifyTime(result.rawCells()[0].getTimestamp());
439. metaData.setLength(len);
440. metaData.setMediaType(Bytes.toString(result.getValue(HosUtil.OBJ_META_CF_BYTES,
441. HosUtil.OBJ_MEDIATYPE_QUALIFIER)));
442. byte[] b = result
443. .getValue(HosUtil.OBJ_META_CF_BYTES, HosUtil.OBJ_PROPS_QUALIFIER);
444. if (b != null) {
445. metaData.setAttrs(JsonUtil.fromJson(Map.class, Bytes.toString(b)));
446. }
447. object.setMetaData(metaData);
448. return object;
449. }
450.
451. @Override
452. public void deleteObject(String bucket, String key) throws Exception {
453. if (key.endsWith("/")) {
454. //check sub dir and current dir files.
455. if (!isDirEmpty(bucket, key)) {
456. throw new RuntimeException("dir is not empty");
457. }
458. InterProcessMutex lock = null;
459. try {
460. String lockey = key.replaceAll("/", "_");
461. lock = new InterProcessMutex(this.zkClient, "/mos/" + bucket + "/" + lockey);
462. lock.acquire();
463. if (!isDirEmpty(bucket, key)) {
464. throw new RuntimeException("dir is not empty");
465. }
466. String dir1 = key.substring(0, key.lastIndexOf("/"));
467. String name = dir1.substring(dir1.lastIndexOf("/") + 1);
468. if (name.length() > 0) {
469. String parent = key.substring(0, key.lastIndexOf(name));
470. HBaseService
471. .deleteQualifier(connection, HosUtil.getDirTableName(bucket), parent,
472. HosUtil.DIR_SUBDIR_CF, name);
473. }
474. HBaseService.delete(connection, HosUtil.getDirTableName(bucket), key);
475. return;
476. } finally {
477. if (lock != null) {
478. lock.release();
479. }
480. }
481. }
482. String dir = key.substring(0, key.lastIndexOf("/") + 1);
483. String name = key.substring(key.lastIndexOf("/") + 1);
484. String seqId = this.getDirSeqId(bucket, dir);
485. String objKey = seqId + "_" + name;
486. Result result = HBaseService
487. .getRow(connection, HosUtil.getObjTableName(bucket), objKey,
488. HosUtil.OBJ_META_CF_BYTES, HosUtil.OBJ_LEN_QUALIFIER);
489. if (result.isEmpty()) {
490. return;
491. }
492. long len = Bytes.toLong(result.getValue(HosUtil.OBJ_META_CF_BYTES,
493. HosUtil.OBJ_LEN_QUALIFIER));
494. if (len > HosUtil.FILE_STORE_THRESHOLD) {
495. String fileDir = HosUtil.FILE_STORE_ROOT + "/" + bucket + "/" + seqId;
496. this.fileStore.deleteFile(fileDir, name);
497. }
498. HBaseService.delete(connection, HosUtil.getObjTableName(bucket), objKey);
499. }
500.
501. private boolean isDirEmpty(String bucket, String dir) throws IOException {
502. return listDir(bucket, dir, null, 2).getObjectList().size() == 0;
503. }
504.
505. @Override
506. public void deleteBucketStore(String bucket) throws IOException {
507. HBaseService.deleteTable(connection, HosUtil.getDirTableName(bucket));
508. HBaseService.deleteTable(connection, HosUtil.getObjTableName(bucket));
509.
510. HBaseService.delete(connection, HosUtil.BUCKET_DIR_SEQ_TABLE, bucket);
511. this.fileStore.deleteDir(HosUtil.FILE_STORE_ROOT + "/" + bucket);
512. }
513.
514. @Override
515. public void createBucketStore(String bucket) throws IOException {
516. HBaseService.createTable(connection, HosUtil.getDirTableName(bucket),
517. HosUtil.getDirColumnFamily());
518. HBaseService.createTable(connection, HosUtil.getObjTableName(bucket),
519. HosUtil.getObjColumnFamily(), HosUtil.OBJ_REGIONS);
520.
521. Put put = new Put(Bytes.toBytes(bucket));
522. put.addColumn(HosUtil.BUCKET_DIR_SEQ_CF_BYTES,
523. HosUtil.BUCKET_DIR_SEQ_QUALIFIER, Bytes.toBytes(0L));
524. HBaseService.putRow(connection, HosUtil.BUCKET_DIR_SEQ_TABLE, put);
525. this.fileStore.mikDir(HosUtil.FILE_STORE_ROOT + "/" + bucket);
526. }
527.
528. /**
529. * seqId由bucket和dir唯一确定
530. * @param bucket
531. * @param dir
532. * @return
533. * @throws IOException
534. */
535. private String getDirSeqId(String bucket, String dir) throws IOException {
536. Result result = HBaseService.getRow(connection, HosUtil.getDirTableName(bucket), dir);
537. if (result.isEmpty()) {
538. return null;
539. }
540. String dirSeqId = Bytes.toString(result.getValue(HosUtil.DIR_META_CF_BYTES,
541. HosUtil.DIR_SEQID_QUALIFIER));
542. return dirSeqId;
543. }
544.
545. private void getDirAllFiles(String bucket, String dir, String seqId, List<HosObjectSummary> keys,
546. String endKey) throws IOException {
547.
548. byte[] max = Bytes.createMaxByteArray(100);
549. byte[] tail = Bytes.add(Bytes.toBytes(seqId), max);
550. if (endKey.startsWith(dir)) {
551. String endKeyLeft = endKey.replace(dir, "");
552. String fileNameMax = endKeyLeft;
553. if (endKeyLeft.indexOf("/") > 0) {
554. fileNameMax = endKeyLeft.substring(0, endKeyLeft.indexOf("/"));
555. }
556. tail = Bytes.toBytes(seqId + "_" + fileNameMax);
557. }
558.
559. Scan scan = new Scan(Bytes.toBytes(seqId), tail);
560. scan.setFilter(HosUtil.OBJ_META_SCAN_FILTER);
561. ResultScanner scanner = HBaseService
562. .scanner(connection, HosUtil.getObjTableName(bucket), scan);
563. Result result = null;
564. while ((result = scanner.next()) != null) {
565. HosObjectSummary summary = this.resultToObjectSummary(result, bucket, dir);
566. keys.add(summary);
567. }
568. if (scanner != null) {
569. scanner.close();
570. }
571. }
572.
573. private HosObjectSummary resultToObjectSummary(Result result, String bucket, String dir)
574. throws IOException {
575. HosObjectSummary summary = new HosObjectSummary();
576. long timestamp = result.rawCells()[0].getTimestamp();
577. summary.setLastModifyTime(timestamp);
578. String id = new String(result.getRow());
579. summary.setId(id);
580. String name = id.split("_", 2)[1];
581. String key = dir + name;
582. summary.setKey(key);
583. summary.setName(name);
584. summary.setBucket(bucket);
585. String s = Bytes.toString(result.getValue(HosUtil.OBJ_META_CF_BYTES,
586. HosUtil.OBJ_PROPS_QUALIFIER));
587. if (s != null) {
588. summary.setAttrs(JsonUtil.fromJson(Map.class, s));
589. }
590. summary.setLength(Bytes.toLong(result.getValue(HosUtil.OBJ_META_CF_BYTES,
591. HosUtil.OBJ_LEN_QUALIFIER)));
592. summary
593. .setMediaType(Bytes.toString(result.getValue(HosUtil.OBJ_META_CF_BYTES,
594. HosUtil.OBJ_MEDIATYPE_QUALIFIER)));
595.
596. return summary;
597. }
598.
599. private HosObjectSummary dirObjectToSummary(Result result, String bucket, String dir) {
600. HosObjectSummary summary = new HosObjectSummary();
601. String id = Bytes.toString(result.getRow());
602. summary.setId(id);
603. summary.setAttrs(new HashMap<>(0));
604. if (dir.length() > 1) {
605. summary.setName(dir.substring(dir.lastIndexOf("/") + 1));
606. } else {
607. summary.setName("");
608. }
609. summary.setBucket(bucket);
610. summary.setKey(dir);
611. summary.setLastModifyTime(result.rawCells()[0].getTimestamp());
612. summary.setLength(0);
613. summary.setMediaType("");
614. return summary;
615. }
616.
617.
618. private short getBucketReplication(String bucket) {
619. return 2;
620. }
621.
622. private String putDir(String bucket, String dir) throws Exception {
623. if (dirExist(bucket, dir)) {
624. return null;
625. }
626. InterProcessMutex lock = null;
627. try {
628. String lockey = dir.replaceAll("/", "_");
629. lock = new InterProcessMutex(this.zkClient, "/mos/" + bucket + "/" + lockey);
630. lock.acquire();
631. String dir1 = dir.substring(0, dir.lastIndexOf("/"));
632. String name = dir1.substring(dir1.lastIndexOf("/") + 1);
633. if (name.length() > 0) {
634. String parent = dir.substring(0, dir1.lastIndexOf("/") + 1);
635. if (!this.dirExist(bucket, parent)) {
636. this.putDir(bucket, parent);
637. }
638. Put put = new Put(Bytes.toBytes(parent));
639. put.addColumn(HosUtil.DIR_SUBDIR_CF_BYTES, Bytes.toBytes(name),
640. Bytes.toBytes('1'));
641. HBaseService.putRow(connection, HosUtil.getDirTableName(bucket), put);
642. }
643. String seqId = this.getDirSeqId(bucket, dir);
644. String hash = seqId == null ? makeDirSeqId(bucket) : seqId;
645. Put hashPut = new Put(dir.getBytes());
646. hashPut.addColumn(HosUtil.DIR_META_CF_BYTES,
647. HosUtil.DIR_SEQID_QUALIFIER, Bytes.toBytes(hash));
648. HBaseService.putRow(connection, HosUtil.getDirTableName(bucket), hashPut);
649. return hash;
650. } finally {
651. if (lock != null) {
652. lock.release();
653. }
654. }
655. }
656.
657. private String makeDirSeqId(String bucket) throws IOException {
658. long v = HBaseService
659. .incrementColumnValue(connection, HosUtil.BUCKET_DIR_SEQ_TABLE, bucket,
660. HosUtil.BUCKET_DIR_SEQ_CF_BYTES, HosUtil.BUCKET_DIR_SEQ_QUALIFIER,
661. 1);
662. return String.format("%da%d", v % 64, v);
663. }
664.
665. private boolean dirExist(String bucket, String dir) throws IOException {
666. return HBaseService.existsRow(connection, HosUtil.getDirTableName(bucket), dir);
667. }
668. }
