src/BodyPipe.h

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