src/BodyPipe.h

   1 /*
   2  * Copyright (C) 1996-2014 The Squid Software Foundation and contributors
   3  *
   4  * Squid software is distributed under GPLv2+ license and includes
   5  * contributions from numerous individuals and organizations.
   6  * Please see the COPYING and CONTRIBUTORS files for details.
   7  */
   8
   9 #ifndef SQUID_BODY_PIPE_H
  10 #define SQUID_BODY_PIPE_H
  11
  12 #include "base/AsyncJob.h"
  13 #include "base/CbcPointer.h"
  14 #include "MemBuf.h"
  15
  16 class BodyPipe;
  17
  18 /** Interface for those who want to produce body content for others.
  19  * BodyProducer is expected to create the BodyPipe.
  20  * One pipe cannot have more than one producer.
  21  */
  22 class BodyProducer: virtual public AsyncJob
  23 {
  24 public:
  25     typedef CbcPointer<BodyProducer> Pointer;
  26
  27     BodyProducer():AsyncJob("BodyProducer") {}
  28     virtual ~BodyProducer() {}
  29
  30     virtual void noteMoreBodySpaceAvailable(RefCount<BodyPipe> bp) = 0;
  31     virtual void noteBodyConsumerAborted(RefCount<BodyPipe> bp) = 0;
  32
  33 protected:
  34     void stopProducingFor(RefCount<BodyPipe> &, bool atEof);
  35 };
  36
  37 /** Interface for those who want to consume body content from others.
  38  * BodyConsumer is expected to register with an existing BodyPipe
  39  * by calling BodyPipe::setConsumer().
  40  * One pipe cannot have more than one consumer.
  41  */
  42 class BodyConsumer: virtual public AsyncJob
  43 {
  44 public:
  45     typedef CbcPointer<BodyConsumer> Pointer;
  46
  47     BodyConsumer():AsyncJob("BodyConsumer") {}
  48     virtual ~BodyConsumer() {}
  49
  50     virtual void noteMoreBodyDataAvailable(RefCount<BodyPipe> bp) = 0;
  51     virtual void noteBodyProductionEnded(RefCount<BodyPipe> bp) = 0;
  52     virtual void noteBodyProducerAborted(RefCount<BodyPipe> bp) = 0;
  53
  54 protected:
  55     void stopConsumingFrom(RefCount<BodyPipe> &);
  56 };
  57
  58 /** Makes raw buffer checkin/checkout interface efficient and exception-safe.
  59  * Either append or consume operations can be performed on a checkedout buffer.
  60  */
  61 class BodyPipeCheckout
  62 {
  63 public:
  64     friend class BodyPipe;
  65
  66 public:
  67     BodyPipeCheckout(BodyPipe &); // checks out
  68     ~BodyPipeCheckout(); // aborts checkout unless checkedIn
  69
  70     void checkIn();
  71
  72 public:
  73     BodyPipe &thePipe;
  74     MemBuf &buf;
  75     const uint64_t offset; // of current content, relative to the body start
  76
  77 protected:
  78     const size_t checkedOutSize;
  79     bool checkedIn;
  80
  81 private:
  82     BodyPipeCheckout(const BodyPipeCheckout &); // prevent copying
  83     BodyPipeCheckout &operator =(const BodyPipeCheckout &); // prevent assignment
  84 };
  85
  86 /** Connects those who produces message body content with those who
  87  * consume it. For example, connects ConnStateData with FtpStateData OR
  88  * ICAPModXact with HttpStateData.
  89  */
  90 class BodyPipe: public RefCountable
  91 {
  92 public:
  93     typedef RefCount<BodyPipe> Pointer;
  94     typedef BodyProducer Producer;
  95     typedef BodyConsumer Consumer;
  96     typedef BodyPipeCheckout Checkout;
  97
  98     enum { MaxCapacity = 64*1024 };
  99
 100     friend class BodyPipeCheckout;
 101
 102 public:
 103     BodyPipe(Producer *aProducer);
 104     ~BodyPipe(); // asserts that producer and consumer are cleared
 105
 106     void setBodySize(uint64_t aSize); // set body size
 107     bool bodySizeKnown() const { return theBodySize >= 0; }
 108     uint64_t bodySize() const;
 109     uint64_t consumedSize() const { return theGetSize; }
 110     uint64_t producedSize() const { return thePutSize; }
 111     bool productionEnded() const { return !theProducer; }
 112
 113     // called by producers
 114     void clearProducer(bool atEof); // aborts or sends eof
 115     size_t putMoreData(const char *buf, size_t size);
 116     bool mayNeedMoreData() const { return !bodySizeKnown() || needsMoreData(); }
 117     bool needsMoreData() const { return bodySizeKnown() && unproducedSize() > 0; }
 118     uint64_t unproducedSize() const; // size of still unproduced data
 119     bool stillProducing(const Producer::Pointer &producer) const { return theProducer == producer; }
 120     void expectProductionEndAfter(uint64_t extraSize); ///< sets or checks body size
 121
 122     // called by consumers
 123     bool setConsumerIfNotLate(const Consumer::Pointer &aConsumer);
 124     void clearConsumer(); // aborts if still piping
 125     void expectNoConsumption(); ///< there will be no more setConsumer() calls
 126     size_t getMoreData(MemBuf &buf);
 127     void consume(size_t size);
 128     bool expectMoreAfter(uint64_t offset) const;
 129     bool exhausted() const; // saw eof/abort and all data consumed
 130     bool stillConsuming(const Consumer::Pointer &consumer) const { return theConsumer == consumer; }
 131
 132     // start or continue consuming when there is no consumer
 133     void enableAutoConsumption();
 134
 135     const MemBuf &buf() const { return theBuf; }
 136
 137     const char *status() const; // for debugging only
 138
 139 protected:
 140     // lower-level interface used by Checkout
 141     MemBuf &checkOut(); // obtain raw buffer
 142     void checkIn(Checkout &checkout); // return updated raw buffer
 143     void undoCheckOut(Checkout &checkout); // undo checkout efffect
 144
 145     void scheduleBodyDataNotification();
 146     void scheduleBodyEndNotification();
 147
 148     // keep counters in sync and notify producer or consumer
 149     void postConsume(size_t size);
 150     void postAppend(size_t size);
 151
 152     void startAutoConsumption(); // delayed start of enabled consumption
 153
 154 private:
 155     int64_t  theBodySize;   // expected total content length, if known
 156     Producer::Pointer theProducer; // content producer, if any
 157     Consumer::Pointer theConsumer; // content consumer, if any
 158
 159     uint64_t thePutSize; // ever-increasing total
 160     uint64_t theGetSize; // ever-increasing total
 161
 162     MemBuf theBuf; // produced but not yet consumed content, if any
 163
 164     bool mustAutoConsume; // consume when there is no consumer
 165     bool abortedConsumption; ///< called BodyProducer::noteBodyConsumerAborted
 166     bool isCheckedOut; // to keep track of checkout violations
 167
 168     CBDATA_CLASS2(BodyPipe);
 169 };
 170
 171 #endif /* SQUID_BODY_PIPE_H */