src/BodyPipe.h

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