source: 3thparty/jupload/src/main/java/wjhk/jupload2/upload/PacketConstructionThread.java @ 3951

Revision 3951, 7.6 KB checked in by alexandrecorreia, 13 years ago (diff)

Ticket #1709 - Adicao de codigo fonte java do componente jupload

Line 
1package wjhk.jupload2.upload;
2
3import java.util.concurrent.BlockingQueue;
4
5import wjhk.jupload2.exception.JUploadException;
6import wjhk.jupload2.policies.UploadPolicy;
7
8/**
9 * @author etienne_sf
10 *
11 */
12public class PacketConstructionThread extends Thread {
13
14    /** The current upload policy. */
15    UploadPolicy uploadPolicy = null;
16
17    /** The thread which globally manages the upload */
18    FileUploadManagerThread fileUploadManagerThread = null;
19
20    /**
21     * The queue where each prepared file are stored, for further processing.
22     * This class picks files here, and post them to the packetQueue.
23     */
24    BlockingQueue<UploadFileData> preparedFileQueue = null;
25
26    /**
27     * The queue where each prepared file will be stored, for further processing
28     */
29    BlockingQueue<UploadFilePacket> packetQueue = null;
30
31    /**
32     * The packet this instance is working on.
33     */
34    UploadFilePacket packetInProgress = null;
35
36    /**
37     * Indicates when the last file has been received. The last file is the
38     * poisonned
39     */
40    boolean lastFileReceived = false;
41
42    PacketConstructionThread(BlockingQueue<UploadFileData> preparedFileQueue,
43            BlockingQueue<UploadFilePacket> packetQueue,
44            FileUploadManagerThread fileUploadManagerThread,
45            UploadPolicy uploadPolicy) {
46        // A thread name is very useful, when debugging...
47        super("PacketConstructionThread");
48
49        this.preparedFileQueue = preparedFileQueue;
50        this.packetQueue = packetQueue;
51        this.fileUploadManagerThread = fileUploadManagerThread;
52        this.uploadPolicy = uploadPolicy;
53
54        // Let's construct the first packet...
55        this.packetInProgress = new UploadFilePacket(this.uploadPolicy);
56    }
57
58    /**
59     * The actual command to generate packets.
60     *
61     * @see java.lang.Thread#run()
62     */
63    @Override
64    final public void run() {
65        this.uploadPolicy.displayDebug("Start of PacketConstructionThread", 80);
66        try { // catch (JUploadException e)
67
68            // We loop, and wait for the 'poisonned' UploadFileData to finish.
69            try {
70                while (!lastFileReceived
71                        && !this.fileUploadManagerThread.isUploadFinished()) {
72                    UploadFileData ufd = preparedFileQueue.take();
73                    receiveNewFile(ufd);
74                }
75
76                this.uploadPolicy
77                        .displayDebug(
78                                "PacketConstructionThread: end of loop, the thread is about to finish",
79                                30);
80
81                // We may have some file left to send...
82                if (this.packetInProgress.size() > 0) {
83                    this.uploadPolicy
84                            .displayDebug(
85                                    "Last file received: the current packet is not empty, we send it",
86                                    30);
87                    sendCurrentPacket();
88                }
89            } catch (InterruptedException e) {
90                this.uploadPolicy
91                        .displayWarn("packetConstructionThread received InterruptedException, exiting");
92            }
93
94            // In standard mode, we should have no more file to manage. The
95            // following test is meaningful only if the FilePreparationThread
96            // has been finished before (otherwise, other files could enter the
97            // queue after this test)
98            if (!this.preparedFileQueue.isEmpty()) {
99                if (!this.fileUploadManagerThread.isUploadFinished()) {
100                    // Hum, hum. This should not happen.
101                    this.uploadPolicy
102                            .displayWarn("The preparedFileQueue is not empty, at the end of "
103                                    + this.getClass().getName());
104                }
105                // This can happen, if we are interrupted while working. Let's
106                // empty this.
107                this.uploadPolicy
108                        .displayDebug(
109                                "The PacketConstructionThread is about to finish, but the preparedFileQueue is not empty. Let's clear it.",
110                                30);
111                while (!this.preparedFileQueue.isEmpty()) {
112                    this.preparedFileQueue.poll();
113                }
114            }
115        } catch (JUploadException e) {
116            this.fileUploadManagerThread.setUploadException(e);
117        } finally {
118            // To properly finish the job, we send a 'poisonned' packet, so that
119            // the FileUploadThread knows it's finished.
120            try {
121                this.packetQueue.put(new UploadFilePacketPoisonned(
122                        this.uploadPolicy));
123            } catch (InterruptedException e) {
124                this.uploadPolicy
125                        .displayWarn("packetConstructionThread received InterruptedException (while checking if packetQueue is empty), exiting");
126            }
127        }
128        this.uploadPolicy.displayDebug("End of PacketConstructionThread", 80);
129    }
130
131    /**
132     * Called when a new file is received
133     *
134     * @param uploadFileData
135     * @throws JUploadException
136     * @throws InterruptedException
137     */
138    private void receiveNewFile(UploadFileData uploadFileData)
139            throws JUploadException, InterruptedException {
140        // Are we finished ?
141        if (uploadFileData.isPoisonned()) {
142            lastFileReceived = true;
143            this.uploadPolicy
144                    .displayDebug(
145                            "Poisonned UploadFileData received, PacketContructionThread will exit normally",
146                            50);
147        } else {
148            // We try to add the file to the current packet. If it doesn't work,
149            // this packet is probably full. We sent it, and add the packet to
150            // the new one.
151            if (!this.packetInProgress.add(uploadFileData)) {
152                // The packet was refused. We send the current one, and retry.
153                this.uploadPolicy
154                        .displayDebug(
155                                "The file can't be added to the current packet. Let's send this packet first.",
156                                80);
157                sendCurrentPacket();
158                if (!this.packetInProgress.add(uploadFileData)) {
159                    throw new JUploadException(
160                            "Could not add file to packet! (filename: "
161                                    + uploadFileData.getFileName() + ")");
162                }
163            }
164
165            // If the current packet is finished, we send it immediatly.
166            if (this.packetInProgress.isFull()) {
167                sendCurrentPacket();
168            }
169        }
170    }
171
172    private void sendCurrentPacket() throws InterruptedException {
173        if (this.packetInProgress == null) {
174            throw new java.lang.AssertionError(
175                    this.getClass().getName()
176                            + ".sendCurrentPacket(): this.packetInProgress may not be null");
177        } else if (this.packetInProgress.size() == 0) {
178            throw new java.lang.AssertionError(
179                    this.getClass().getName()
180                            + ".sendCurrentPacket(): this.packetInProgress.size() may not be 0");
181        }
182        // If a packet is ready, we post it into the relevant queue.
183        this.packetQueue.put(packetInProgress);
184
185        // And we start a new one.
186        this.packetInProgress = new UploadFilePacket(this.uploadPolicy);
187    }
188}
Note: See TracBrowser for help on using the repository browser.