1 #ifndef SQUID_BODY_PIPE_H
2 #define SQUID_BODY_PIPE_H
5 #include "base/AsyncJob.h"
6 #include "base/CbcPointer.h"
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.
14 class BodyProducer
: virtual public AsyncJob
17 typedef CbcPointer
<BodyProducer
> Pointer
;
19 BodyProducer():AsyncJob("BodyProducer") {}
20 virtual ~BodyProducer() {}
22 virtual void noteMoreBodySpaceAvailable(RefCount
<BodyPipe
> bp
) = 0;
23 virtual void noteBodyConsumerAborted(RefCount
<BodyPipe
> bp
) = 0;
26 void stopProducingFor(RefCount
<BodyPipe
> &pipe
, bool atEof
);
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.
34 class BodyConsumer
: virtual public AsyncJob
37 typedef CbcPointer
<BodyConsumer
> Pointer
;
39 BodyConsumer():AsyncJob("BodyConsumer") {}
40 virtual ~BodyConsumer() {}
42 virtual void noteMoreBodyDataAvailable(RefCount
<BodyPipe
> bp
) = 0;
43 virtual void noteBodyProductionEnded(RefCount
<BodyPipe
> bp
) = 0;
44 virtual void noteBodyProducerAborted(RefCount
<BodyPipe
> bp
) = 0;
47 void stopConsumingFrom(RefCount
<BodyPipe
> &pipe
);
50 /** Makes raw buffer checkin/checkout interface efficient and exception-safe.
51 * Either append or consume operations can be performed on a checkedout buffer.
53 class BodyPipeCheckout
56 friend class BodyPipe
;
59 BodyPipeCheckout(BodyPipe
&pipe
); // checks out
60 ~BodyPipeCheckout(); // aborts checkout unless checkedIn
67 const uint64_t offset
; // of current content, relative to the body start
70 const size_t checkedOutSize
;
74 BodyPipeCheckout(const BodyPipeCheckout
&); // prevent copying
75 BodyPipeCheckout
&operator =(const BodyPipeCheckout
&); // prevent assignment
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.
82 class BodyPipe
: public RefCountable
85 typedef RefCount
<BodyPipe
> Pointer
;
86 typedef BodyProducer Producer
;
87 typedef BodyConsumer Consumer
;
88 typedef BodyPipeCheckout Checkout
;
90 enum { MaxCapacity
= SQUID_TCP_SO_RCVBUF
};
92 friend class BodyPipeCheckout
;
95 BodyPipe(Producer
*aProducer
);
96 ~BodyPipe(); // asserts that producer and consumer are cleared
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 bool productionEnded() const { return !theProducer
; }
104 // called by producers
105 void clearProducer(bool atEof
); // aborts or sends eof
106 size_t putMoreData(const char *buf
, size_t size
);
107 bool mayNeedMoreData() const { return !bodySizeKnown() || needsMoreData(); }
108 bool needsMoreData() const { return bodySizeKnown() && unproducedSize() > 0; }
109 uint64_t unproducedSize() const; // size of still unproduced data
110 bool stillProducing(const Producer::Pointer
&producer
) const { return theProducer
== producer
; }
111 void expectProductionEndAfter(uint64_t extraSize
); ///< sets or checks body size
113 // called by consumers
114 bool setConsumerIfNotLate(const Consumer::Pointer
&aConsumer
);
115 void clearConsumer(); // aborts if still piping
116 size_t getMoreData(MemBuf
&buf
);
117 void consume(size_t size
);
118 bool expectMoreAfter(uint64_t offset
) const;
119 bool exhausted() const; // saw eof/abort and all data consumed
120 bool stillConsuming(const Consumer::Pointer
&consumer
) const { return theConsumer
== consumer
; }
122 // start or continue consuming when there is no consumer
123 void enableAutoConsumption();
125 const MemBuf
&buf() const { return theBuf
; }
127 const char *status() const; // for debugging only
130 // lower-level interface used by Checkout
131 MemBuf
&checkOut(); // obtain raw buffer
132 void checkIn(Checkout
&checkout
); // return updated raw buffer
133 void undoCheckOut(Checkout
&checkout
); // undo checkout efffect
135 void scheduleBodyDataNotification();
136 void scheduleBodyEndNotification();
138 // keep counters in sync and notify producer or consumer
139 void postConsume(size_t size
);
140 void postAppend(size_t size
);
142 void startAutoConsumption(); // delayed start of enabled consumption
145 int64_t theBodySize
; // expected total content length, if known
146 Producer::Pointer theProducer
; // content producer, if any
147 Consumer::Pointer theConsumer
; // content consumer, if any
149 uint64_t thePutSize
; // ever-increasing total
150 uint64_t theGetSize
; // ever-increasing total
152 MemBuf theBuf
; // produced but not yet consumed content, if any
154 bool mustAutoConsume
; // consume when there is no consumer
155 bool isCheckedOut
; // to keep track of checkout violations
157 CBDATA_CLASS2(BodyPipe
);
160 #endif /* SQUID_BODY_PIPE_H */