001/* ***** BEGIN LICENSE BLOCK *****
002 * Version: MPL 1.1/GPL 2.0/LGPL 2.1
003 *
004 * The contents of this file are subject to the Mozilla Public License Version
005 * 1.1 (the "License"); you may not use this file except in compliance with
006 * the License. You may obtain a copy of the License at
007 * http://www.mozilla.org/MPL/
008 *
009 * Software distributed under the License is distributed on an "AS IS" basis,
010 * WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
011 * for the specific language governing rights and limitations under the
012 * License.
013 *
014 * The Original Code is part of dcm4che, an implementation of DICOM(TM) in
015 * Java(TM), hosted at https://github.com/gunterze/dcm4che.
016 *
017 * The Initial Developer of the Original Code is
018 * Agfa Healthcare.
019 * Portions created by the Initial Developer are Copyright (C) 2011
020 * the Initial Developer. All Rights Reserved.
021 *
022 * Contributor(s):
023 * See @authors listed below
024 *
025 * Alternatively, the contents of this file may be used under the terms of
026 * either the GNU General Public License Version 2 or later (the "GPL"), or
027 * the GNU Lesser General Public License Version 2.1 or later (the "LGPL"),
028 * in which case the provisions of the GPL or the LGPL are applicable instead
029 * of those above. If you wish to allow use of your version of this file only
030 * under the terms of either the GPL or the LGPL, and not to allow others to
031 * use your version of this file under the terms of the MPL, indicate your
032 * decision by deleting the provisions above and replace them with the notice
033 * and other provisions required by the GPL or the LGPL. If you do not delete
034 * the provisions above, a recipient may use your version of this file under
035 * the terms of any one of the MPL, the GPL or the LGPL.
036 *
037 * ***** END LICENSE BLOCK ***** */
038
039package org.dcm4che3.net.service;
040
041import java.io.IOException;
042import java.util.ArrayList;
043import java.util.Iterator;
044import java.util.List;
045import java.util.Observable;
046import java.util.Observer;
047import java.util.concurrent.ScheduledFuture;
048import java.util.concurrent.TimeUnit;
049
050import org.dcm4che3.data.Attributes;
051import org.dcm4che3.data.Tag;
052import org.dcm4che3.data.UID;
053import org.dcm4che3.data.VR;
054import org.dcm4che3.io.DicomInputStream;
055import org.dcm4che3.net.Association;
056import org.dcm4che3.net.Commands;
057import org.dcm4che3.net.DataWriter;
058import org.dcm4che3.net.Dimse;
059import org.dcm4che3.net.DimseRSPHandler;
060import org.dcm4che3.net.InputStreamDataWriter;
061import org.dcm4che3.net.Status;
062import org.dcm4che3.net.pdu.PresentationContext;
063import org.slf4j.Logger;
064import org.slf4j.LoggerFactory;
065
066/**
067 * @author Gunter Zeilinger <gunterze@gmail.com>
068 * @author Umberto Cappellini <umberto.cappellini@agfa.com>
069 */
070public class BasicRetrieveTask<T extends InstanceLocator> implements
071        RetrieveTask {
072
073    protected static final Logger LOG = LoggerFactory
074            .getLogger(BasicRetrieveTask.class);
075
076    protected final List<T> insts;
077    protected final Dimse rq;
078    protected final Association rqas;
079    protected final Association storeas;
080    protected final PresentationContext pc;
081    protected final Attributes rqCmd;
082    protected final int msgId;
083    protected final int priority;
084    protected boolean pendingRSP;
085    protected int pendingRSPInterval;
086    protected int outstandingRSP = 0;
087    protected Object outstandingRSPLock = new Object();
088
089    private CStoreSCU<T> storescu;
090    private ScheduledFuture<?> writePendingRSP;
091
092    public BasicRetrieveTask(Dimse rq, Association rqas,
093            PresentationContext pc, Attributes rqCmd, List<T> insts,
094            Association storeas, CStoreSCU<T> storescu) {
095        this.rq = rq;
096        this.rqas = rqas;
097        this.storeas = storeas;
098        this.pc = pc;
099        this.rqCmd = rqCmd;
100        this.insts = insts;
101        this.msgId = rqCmd.getInt(Tag.MessageID, -1);
102        this.priority = rqCmd.getInt(Tag.Priority, 0);
103        this.storescu = storescu;
104    }
105
106    public void setSendPendingRSP(boolean pendingRSP) {
107        this.pendingRSP = pendingRSP;
108    }
109
110    public void setSendPendingRSPInterval(int pendingRSPInterval) {
111        this.pendingRSPInterval = pendingRSPInterval;
112    }
113
114    public boolean isCMove() {
115        return rq == Dimse.C_MOVE_RQ;
116    }
117
118    public Association getRequestAssociation() {
119        return rqas;
120    }
121
122    public Association getStoreAssociation() {
123        return storeas;
124    }
125
126    @Override
127    public void onCancelRQ(Association as) {
128        storescu.cancel();
129    }
130
131    @Override
132    public void run() {
133        rqas.addCancelRQHandler(msgId, this);
134        ((Observable)storescu).addObserver(this);
135        try {
136            if (pendingRSPInterval > 0)
137                startWritingAsyncRSP();
138            storescu.cstore(insts, storeas, priority);
139            if (isCMove())
140                releaseStoreAssociation(storeas);
141            stopWritingAsyncRSP();
142            writeRSP(); //last response
143        } finally {
144            rqas.removeCancelRQHandler(msgId);
145            try {
146                close();
147            } catch (Throwable e) {
148                LOG.warn("Exception thrown by {}.close()",
149                        getClass().getName(), e);
150            }
151        }
152    }
153
154    protected void releaseStoreAssociation(Association storeas) {
155        try {
156            storeas.release();
157        } catch (IOException e) {
158            LOG.warn("{}: failed to release association to {}", rqas,
159                    storeas.getRemoteAET(), e);
160        }
161    }
162
163    private void startWritingAsyncRSP() {
164        writePendingRSP = rqas.getApplicationEntity().getDevice()
165                .scheduleAtFixedRate(new Runnable() {
166                    @Override
167                    public void run() {
168                        BasicRetrieveTask.this.writeRSP(); // async response
169                    }
170                }, 0, pendingRSPInterval, TimeUnit.SECONDS);
171    }
172
173    private void stopWritingAsyncRSP() {
174        if (writePendingRSP != null) {
175            writePendingRSP.cancel(false);
176        }
177    }
178
179    private void writeRSP() {
180        try {
181
182            Attributes cmd = Commands.mkRSP(rqCmd, storescu.getStatus(), rq);
183            if (storescu.getStatus() == Status.Pending
184                    || storescu.getStatus() == Status.Cancel)
185                cmd.setInt(Tag.NumberOfRemainingSuboperations, VR.US,
186                        storescu.getRemaining());
187            cmd.setInt(Tag.NumberOfCompletedSuboperations, VR.US, storescu
188                    .getCompleted().size());
189            cmd.setInt(Tag.NumberOfFailedSuboperations, VR.US, storescu
190                    .getFailed().size());
191            cmd.setInt(Tag.NumberOfWarningSuboperations, VR.US, storescu
192                    .getWarning().size());
193            Attributes data = null;
194            if (!storescu.getFailed().isEmpty()
195                    && storescu.getStatus() != Status.Pending) {
196                data = new Attributes(1);
197                String[] iuids = new String[storescu.getFailed().size()];
198                for (int i = 0; i < iuids.length; i++) {
199                    iuids[i] = storescu.getFailed().get(i).iuid;
200                }
201                data.setString(Tag.FailedSOPInstanceUIDList, VR.UI, iuids);
202            }
203            rqas.writeDimseRSP(pc, cmd, data);
204
205        } catch (IOException e) {
206            pendingRSP = false;
207            stopWritingAsyncRSP();
208            LOG.warn(
209                    "{}: Unable to send C-GET or C-MOVE RSP on association to {}",
210                    rqas, rqas.getRemoteAET(), e);
211        }
212    }
213
214    protected void close() {
215    }
216
217    // notification from cstorescu
218    public void update(Observable obj, Object arg) {
219
220        storescu = (CStoreSCU<T>) obj;
221
222        if (pendingRSP && storescu.getStatus() == Status.Pending)
223            writeRSP(); // sync response
224    }
225
226}