src/BodyPipe.h

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