src/BodyPipe.h

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