1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
|
package org.simpleframework.common.buffer;
import java.io.IOException;
import java.io.InputStream;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import org.simpleframework.common.buffer.Allocator;
import org.simpleframework.common.buffer.Buffer;
import org.simpleframework.common.buffer.BufferAllocator;
public class FileByteQueue {
private BlockingQueue<Block> blocks;
private BlockAllocator allocator;
private Block source;
public FileByteQueue(Allocator allocator) throws IOException {
this.blocks = new LinkedBlockingQueue<Block>();
this.allocator = new BlockAllocator(allocator);
}
public int read(byte[] array, int off, int size) throws Exception {
int left = blocks.size();
int mark = size;
for(int i = 0; source != null || i < left; i++) {
if(source == null) {
source = blocks.take();
}
int remain = source.remaining();
int read = Math.min(remain, size);
if(read > 0) {
source.read(array, off, size);
size -= read;
off += read;
}
if(remain == 0) {
source.close(); // clear up file handles
source = null;
}
if(size <= 0) {
return mark;
}
}
return mark - size;
}
public void write(byte[] array, int off, int size) throws Exception {
Block buffer = allocator.allocate(array, off, size);
if(size > 0) {
blocks.offer(buffer);
}
}
private class BlockAllocator {
private Allocator allocator;
public BlockAllocator(Allocator allocator) {
this.allocator = new BufferAllocator(allocator);
}
public Block allocate(byte[] array, int off, int size) throws IOException {
Buffer buffer = allocator.allocate();
if(size > 0) {
buffer.append(array, off, size);
}
return new Block(buffer, size);
}
}
private class Block {
private InputStream source;
private int remaining;
private int size;
public Block(Buffer buffer, int size) throws IOException {
this.source = buffer.open();
this.remaining = size;
this.size = size;
}
public int read(byte[] array, int off, int size) throws IOException {
int count = source.read(array, off, size);
if(count > 0) {
remaining -= size;
}
return count;
}
public void close() throws IOException {
source.close();
}
public int remaining() {
return remaining;
}
public int size() {
return size;
}
}
}
|
