src/adaptation/ecap/XactionRep.cc

   1 /*
   2  * Copyright (C) 1996-2018 The Squid Software Foundation and contributors
   3  *
   4  * Squid software is distributed under GPLv2+ license and includes
   5  * contributions from numerous individuals and organizations.
   6  * Please see the COPYING and CONTRIBUTORS files for details.
   7  */
   8
   9 /* DEBUG: section 93    eCAP Interface */
  10
  11 #include "squid.h"
  12 #include <libecap/common/area.h>
  13 #include <libecap/common/delay.h>
  14 #include <libecap/common/named_values.h>
  15 #include <libecap/common/names.h>
  16 #include <libecap/adapter/xaction.h>
  17 #include "adaptation/Answer.h"
  18 #include "adaptation/ecap/Config.h"
  19 #include "adaptation/ecap/XactionRep.h"
  20 #include "adaptation/Initiator.h"
  21 #include "base/AsyncJobCalls.h"
  22 #include "base/TextException.h"
  23 #include "format/Format.h"
  24 #include "HttpReply.h"
  25 #include "HttpRequest.h"
  26 #include "MasterXaction.h"
  27 #include "SquidTime.h"
  28
  29 CBDATA_NAMESPACED_CLASS_INIT(Adaptation::Ecap::XactionRep, XactionRep);
  30
  31 /// a libecap Visitor for converting adapter transaction options to HttpHeader
  32 class OptionsExtractor: public libecap::NamedValueVisitor
  33 {
  34 public:
  35     typedef libecap::Name Name;
  36     typedef libecap::Area Area;
  37
  38     OptionsExtractor(HttpHeader &aMeta): meta(aMeta) {}
  39
  40     // libecap::NamedValueVisitor API
  41     virtual void visit(const Name &name, const Area &value) {
  42         meta.putExt(name.image().c_str(), value.toString().c_str());
  43     }
  44
  45     HttpHeader &meta; ///< where to put extracted options
  46 };
  47
  48 Adaptation::Ecap::XactionRep::XactionRep(
  49     Http::Message *virginHeader, HttpRequest *virginCause, AccessLogEntry::Pointer &alp,
  50     const Adaptation::ServicePointer &aService):
  51     AsyncJob("Adaptation::Ecap::XactionRep"),
  52     Adaptation::Initiate("Adaptation::Ecap::XactionRep"),
  53     theService(aService),
  54     theVirginRep(virginHeader), theCauseRep(NULL),
  55     makingVb(opUndecided), proxyingAb(opUndecided),
  56     adaptHistoryId(-1),
  57     vbProductionFinished(false),
  58     abProductionFinished(false), abProductionAtEnd(false),
  59     al(alp)
  60 {
  61     if (virginCause)
  62         theCauseRep = new MessageRep(virginCause);
  63 }
  64
  65 Adaptation::Ecap::XactionRep::~XactionRep()
  66 {
  67     assert(!theMaster);
  68     delete theCauseRep;
  69     theAnswerRep.reset();
  70 }
  71
  72 void
  73 Adaptation::Ecap::XactionRep::master(const AdapterXaction &x)
  74 {
  75     Must(!theMaster);
  76     Must(x);
  77     theMaster = x;
  78 }
  79
  80 Adaptation::Service &
  81 Adaptation::Ecap::XactionRep::service()
  82 {
  83     Must(theService != NULL);
  84     return *theService;
  85 }
  86
  87 const libecap::Area
  88 Adaptation::Ecap::XactionRep::option(const libecap::Name &name) const
  89 {
  90     if (name == libecap::metaClientIp)
  91         return clientIpValue();
  92     if (name == libecap::metaUserName)
  93         return usernameValue();
  94     if (Adaptation::Config::masterx_shared_name && name == Adaptation::Config::masterx_shared_name)
  95         return masterxSharedValue(name);
  96
  97     // TODO: metaServerIp, metaAuthenticatedUser, and metaAuthenticatedGroups
  98
  99     // If the name is unknown, metaValue returns an emtpy area
 100     return metaValue(name);
 101 }
 102
 103 void
 104 Adaptation::Ecap::XactionRep::visitEachOption(libecap::NamedValueVisitor &visitor) const
 105 {
 106     if (const libecap::Area value = clientIpValue())
 107         visitor.visit(libecap::metaClientIp, value);
 108     if (const libecap::Area value = usernameValue())
 109         visitor.visit(libecap::metaUserName, value);
 110
 111     if (Adaptation::Config::masterx_shared_name) {
 112         const libecap::Name name(Adaptation::Config::masterx_shared_name);
 113         if (const libecap::Area value = masterxSharedValue(name))
 114             visitor.visit(name, value);
 115     }
 116
 117     visitEachMetaHeader(visitor);
 118
 119     // TODO: metaServerIp, metaAuthenticatedUser, and metaAuthenticatedGroups
 120 }
 121
 122 const libecap::Area
 123 Adaptation::Ecap::XactionRep::clientIpValue() const
 124 {
 125     const HttpRequest *request = dynamic_cast<const HttpRequest*>(theCauseRep ?
 126                                  theCauseRep->raw().header : theVirginRep.raw().header);
 127     Must(request);
 128     // TODO: move this logic into HttpRequest::clientIp(bool) and
 129     // HttpRequest::clientIpString(bool) and reuse everywhere
 130     if (TheConfig.send_client_ip && request) {
 131         Ip::Address client_addr;
 132 #if FOLLOW_X_FORWARDED_FOR
 133         if (TheConfig.use_indirect_client) {
 134             client_addr = request->indirect_client_addr;
 135         } else
 136 #endif
 137             client_addr = request->client_addr;
 138         if (!client_addr.isAnyAddr() && !client_addr.isNoAddr()) {
 139             char ntoabuf[MAX_IPSTRLEN] = "";
 140             client_addr.toStr(ntoabuf,MAX_IPSTRLEN);
 141             return libecap::Area::FromTempBuffer(ntoabuf, strlen(ntoabuf));
 142         }
 143     }
 144     return libecap::Area();
 145 }
 146
 147 const libecap::Area
 148 Adaptation::Ecap::XactionRep::usernameValue() const
 149 {
 150 #if USE_AUTH
 151     const HttpRequest *request = dynamic_cast<const HttpRequest*>(theCauseRep ?
 152                                  theCauseRep->raw().header : theVirginRep.raw().header);
 153     Must(request);
 154     if (request->auth_user_request != NULL) {
 155         if (char const *name = request->auth_user_request->username())
 156             return libecap::Area::FromTempBuffer(name, strlen(name));
 157         else if (request->extacl_user.size() > 0)
 158             return libecap::Area::FromTempBuffer(request->extacl_user.rawBuf(),
 159                                                  request->extacl_user.size());
 160     }
 161 #endif
 162     return libecap::Area();
 163 }
 164
 165 const libecap::Area
 166 Adaptation::Ecap::XactionRep::masterxSharedValue(const libecap::Name &name) const
 167 {
 168     const HttpRequest *request = dynamic_cast<const HttpRequest*>(theCauseRep ?
 169                                  theCauseRep->raw().header : theVirginRep.raw().header);
 170     Must(request);
 171     if (name.known()) { // must check to avoid empty names matching unset cfg
 172         Adaptation::History::Pointer ah = request->adaptHistory(false);
 173         if (ah != NULL) {
 174             String name, value;
 175             if (ah->getXxRecord(name, value))
 176                 return libecap::Area::FromTempBuffer(value.rawBuf(), value.size());
 177         }
 178     }
 179     return libecap::Area();
 180 }
 181
 182 const libecap::Area
 183 Adaptation::Ecap::XactionRep::metaValue(const libecap::Name &name) const
 184 {
 185     HttpRequest *request = dynamic_cast<HttpRequest*>(theCauseRep ?
 186                            theCauseRep->raw().header : theVirginRep.raw().header);
 187     Must(request);
 188     HttpReply *reply = dynamic_cast<HttpReply*>(theVirginRep.raw().header);
 189
 190     if (name.known()) { // must check to avoid empty names matching unset cfg
 191         typedef Notes::iterator ACAMLI;
 192         for (auto h: Adaptation::Config::metaHeaders) {
 193             if (name == h->key().toStdString()) {
 194                 SBuf matched;
 195                 if (h->match(request, reply, al, matched))
 196                     return libecap::Area::FromTempString(matched.toStdString());
 197                 else
 198                     return libecap::Area();
 199             }
 200         }
 201     }
 202
 203     return libecap::Area();
 204 }
 205
 206 void
 207 Adaptation::Ecap::XactionRep::visitEachMetaHeader(libecap::NamedValueVisitor &visitor) const
 208 {
 209     HttpRequest *request = dynamic_cast<HttpRequest*>(theCauseRep ?
 210                            theCauseRep->raw().header : theVirginRep.raw().header);
 211     Must(request);
 212     HttpReply *reply = dynamic_cast<HttpReply*>(theVirginRep.raw().header);
 213
 214     for (auto h: Adaptation::Config::metaHeaders) {
 215         SBuf matched;
 216         if (h->match(request, reply, al, matched)) {
 217             const libecap::Name name(h->key().toStdString());
 218             const libecap::Area value = libecap::Area::FromTempString(matched.toStdString());
 219             visitor.visit(name, value);
 220         }
 221     }
 222 }
 223
 224 void
 225 Adaptation::Ecap::XactionRep::start()
 226 {
 227     Must(theMaster);
 228
 229     if (!theVirginRep.raw().body_pipe)
 230         makingVb = opNever; // there is nothing to deliver
 231
 232     HttpRequest *request = dynamic_cast<HttpRequest*> (theCauseRep ?
 233                            theCauseRep->raw().header : theVirginRep.raw().header);
 234     Must(request);
 235
 236     HttpReply *reply = dynamic_cast<HttpReply*>(theVirginRep.raw().header);
 237
 238     Adaptation::History::Pointer ah = request->adaptLogHistory();
 239     if (ah != NULL) {
 240         // retrying=false because ecap never retries transactions
 241         adaptHistoryId = ah->recordXactStart(service().cfg().key, current_time, false);
 242         SBuf matched;
 243         for (auto h: Adaptation::Config::metaHeaders) {
 244             if (h->match(request, reply, al, matched)) {
 245                 if (ah->metaHeaders == NULL)
 246                     ah->metaHeaders = new NotePairs();
 247                 if (!ah->metaHeaders->hasPair(h->key(), matched))
 248                     ah->metaHeaders->add(h->key(), matched);
 249             }
 250         }
 251     }
 252
 253     theMaster->start();
 254 }
 255
 256 void
 257 Adaptation::Ecap::XactionRep::swanSong()
 258 {
 259     // clear body_pipes, if any
 260     // this code does not maintain proxying* and canAccessVb states; should it?
 261
 262     if (theAnswerRep) {
 263         BodyPipe::Pointer body_pipe = answer().body_pipe;
 264         if (body_pipe != NULL) {
 265             Must(body_pipe->stillProducing(this));
 266             stopProducingFor(body_pipe, false);
 267         }
 268     }
 269
 270     BodyPipe::Pointer &body_pipe = theVirginRep.raw().body_pipe;
 271     if (body_pipe != NULL && body_pipe->stillConsuming(this))
 272         stopConsumingFrom(body_pipe);
 273
 274     terminateMaster();
 275
 276     const HttpRequest *request = dynamic_cast<const HttpRequest*>(theCauseRep ?
 277                                  theCauseRep->raw().header : theVirginRep.raw().header);
 278     Must(request);
 279     Adaptation::History::Pointer ah = request->adaptLogHistory();
 280     if (ah != NULL && adaptHistoryId >= 0)
 281         ah->recordXactFinish(adaptHistoryId);
 282
 283     Adaptation::Initiate::swanSong();
 284 }
 285
 286 void
 287 Adaptation::Ecap::XactionRep::resume()
 288 {
 289     // go async to gain exception protection and done()-based job destruction
 290     typedef NullaryMemFunT<Adaptation::Ecap::XactionRep> Dialer;
 291     AsyncCall::Pointer call = asyncCall(93, 5, "Adaptation::Ecap::XactionRep::doResume",
 292                                         Dialer(this, &Adaptation::Ecap::XactionRep::doResume));
 293     ScheduleCallHere(call);
 294 }
 295
 296 /// the guts of libecap::host::Xaction::resume() API implementation
 297 /// which just goes async in Adaptation::Ecap::XactionRep::resume().
 298 void
 299 Adaptation::Ecap::XactionRep::doResume()
 300 {
 301     Must(theMaster);
 302     theMaster->resume();
 303 }
 304
 305 libecap::Message &
 306 Adaptation::Ecap::XactionRep::virgin()
 307 {
 308     return theVirginRep;
 309 }
 310
 311 const libecap::Message &
 312 Adaptation::Ecap::XactionRep::cause()
 313 {
 314     Must(theCauseRep != NULL);
 315     return *theCauseRep;
 316 }
 317
 318 libecap::Message &
 319 Adaptation::Ecap::XactionRep::adapted()
 320 {
 321     Must(theAnswerRep);
 322     return *theAnswerRep;
 323 }
 324
 325 Adaptation::Message &
 326 Adaptation::Ecap::XactionRep::answer()
 327 {
 328     MessageRep *rep = dynamic_cast<MessageRep*>(theAnswerRep.get());
 329     Must(rep);
 330     return rep->raw();
 331 }
 332
 333 void
 334 Adaptation::Ecap::XactionRep::terminateMaster()
 335 {
 336     if (theMaster) {
 337         AdapterXaction x = theMaster;
 338         theMaster.reset();
 339         x->stop();
 340     }
 341 }
 342
 343 bool
 344 Adaptation::Ecap::XactionRep::doneAll() const
 345 {
 346     return makingVb >= opComplete && proxyingAb >= opComplete &&
 347            Adaptation::Initiate::doneAll();
 348 }
 349
 350 // stops receiving virgin and enables auto-consumption, dropping any vb bytes
 351 void
 352 Adaptation::Ecap::XactionRep::sinkVb(const char *reason)
 353 {
 354     debugs(93,4, HERE << "sink for " << reason << "; status:" << status());
 355
 356     // we reset raw().body_pipe when we are done, so use this one for checking
 357     const BodyPipePointer &permPipe = theVirginRep.raw().header->body_pipe;
 358     if (permPipe != NULL)
 359         permPipe->enableAutoConsumption();
 360
 361     forgetVb(reason);
 362 }
 363
 364 // stops receiving virgin but preserves it for others to use
 365 void
 366 Adaptation::Ecap::XactionRep::preserveVb(const char *reason)
 367 {
 368     debugs(93,4, HERE << "preserve for " << reason << "; status:" << status());
 369
 370     // we reset raw().body_pipe when we are done, so use this one for checking
 371     const BodyPipePointer &permPipe = theVirginRep.raw().header->body_pipe;
 372     if (permPipe != NULL) {
 373         // if libecap consumed, we cannot preserve
 374         Must(!permPipe->consumedSize());
 375     }
 376
 377     forgetVb(reason);
 378 }
 379
 380 // disassociates us from vb; the last step of sinking or preserving vb
 381 void
 382 Adaptation::Ecap::XactionRep::forgetVb(const char *reason)
 383 {
 384     debugs(93,9, HERE << "forget vb " << reason << "; status:" << status());
 385
 386     BodyPipePointer &p = theVirginRep.raw().body_pipe;
 387     if (p != NULL && p->stillConsuming(this))
 388         stopConsumingFrom(p);
 389
 390     if (makingVb == opUndecided)
 391         makingVb = opNever;
 392     else if (makingVb == opOn)
 393         makingVb = opComplete;
 394 }
 395
 396 void
 397 Adaptation::Ecap::XactionRep::useVirgin()
 398 {
 399     debugs(93,3, HERE << status());
 400     Must(proxyingAb == opUndecided);
 401     proxyingAb = opNever;
 402
 403     preserveVb("useVirgin");
 404
 405     Http::Message *clone = theVirginRep.raw().header->clone();
 406     // check that clone() copies the pipe so that we do not have to
 407     Must(!theVirginRep.raw().header->body_pipe == !clone->body_pipe);
 408
 409     updateHistory(clone);
 410     sendAnswer(Answer::Forward(clone));
 411     Must(done());
 412 }
 413
 414 void
 415 Adaptation::Ecap::XactionRep::useAdapted(const libecap::shared_ptr<libecap::Message> &m)
 416 {
 417     debugs(93,3, HERE << status());
 418     Must(m);
 419     theAnswerRep = m;
 420     Must(proxyingAb == opUndecided);
 421
 422     Http::Message *msg = answer().header;
 423     updateSources(msg);
 424     if (!theAnswerRep->body()) { // final, bodyless answer
 425         proxyingAb = opNever;
 426         updateHistory(msg);
 427         sendAnswer(Answer::Forward(msg));
 428     } else { // got answer headers but need to handle body
 429         proxyingAb = opOn;
 430         Must(!msg->body_pipe); // only host can set body pipes
 431         MessageRep *rep = dynamic_cast<MessageRep*>(theAnswerRep.get());
 432         Must(rep);
 433         rep->tieBody(this); // sets us as a producer
 434         Must(msg->body_pipe != NULL); // check tieBody
 435
 436         updateHistory(msg);
 437         sendAnswer(Answer::Forward(msg));
 438
 439         debugs(93,4, HERE << "adapter will produce body" << status());
 440         theMaster->abMake(); // libecap will produce
 441     }
 442 }
 443
 444 void
 445 Adaptation::Ecap::XactionRep::blockVirgin()
 446 {
 447     debugs(93,3, HERE << status());
 448     Must(proxyingAb == opUndecided);
 449     proxyingAb = opNever;
 450
 451     sinkVb("blockVirgin");
 452
 453     updateHistory(NULL);
 454     sendAnswer(Answer::Block(service().cfg().key));
 455     Must(done());
 456 }
 457
 458 /// Called just before sendAnswer() to record adapter meta-information
 459 /// which may affect answer processing and may be needed for logging.
 460 void
 461 Adaptation::Ecap::XactionRep::updateHistory(Http::Message *adapted)
 462 {
 463     if (!theMaster) // all updates rely on being able to query the adapter
 464         return;
 465
 466     const HttpRequest *request = dynamic_cast<const HttpRequest*>(theCauseRep ?
 467                                  theCauseRep->raw().header : theVirginRep.raw().header);
 468     Must(request);
 469
 470     // TODO: move common ICAP/eCAP logic to Adaptation::Xaction or similar
 471     // TODO: optimize Area-to-String conversion
 472
 473     // update the cross-transactional database if needed
 474     if (const char *xxNameStr = Adaptation::Config::masterx_shared_name) {
 475         Adaptation::History::Pointer ah = request->adaptHistory(true);
 476         if (ah != NULL) {
 477             libecap::Name xxName(xxNameStr); // TODO: optimize?
 478             if (const libecap::Area val = theMaster->option(xxName))
 479                 ah->updateXxRecord(xxNameStr, val.toString().c_str());
 480         }
 481     }
 482
 483     // update the adaptation plan if needed
 484     if (service().cfg().routing) {
 485         String services;
 486         if (const libecap::Area services = theMaster->option(libecap::metaNextServices)) {
 487             Adaptation::History::Pointer ah = request->adaptHistory(true);
 488             if (ah != NULL)
 489                 ah->updateNextServices(services.toString().c_str());
 490         }
 491     } // TODO: else warn (occasionally!) if we got libecap::metaNextServices
 492
 493     // Store received meta headers for adapt::<last_h logformat code use.
 494     // If we already have stored headers from a previous adaptation transaction
 495     // related to the same master transction, they will be replaced.
 496     Adaptation::History::Pointer ah = request->adaptLogHistory();
 497     if (ah != NULL) {
 498         HttpHeader meta(hoReply);
 499         OptionsExtractor extractor(meta);
 500         theMaster->visitEachOption(extractor);
 501         ah->recordMeta(&meta);
 502     }
 503
 504     // Add just-created history to the adapted/cloned request that lacks it.
 505     if (HttpRequest *adaptedReq = dynamic_cast<HttpRequest*>(adapted))
 506         adaptedReq->adaptHistoryImport(*request);
 507 }
 508
 509 void
 510 Adaptation::Ecap::XactionRep::vbDiscard()
 511 {
 512     Must(makingVb == opUndecided);
 513     // if adapter does not need vb, we do not need to send it
 514     sinkVb("vbDiscard");
 515     Must(makingVb == opNever);
 516 }
 517
 518 void
 519 Adaptation::Ecap::XactionRep::vbMake()
 520 {
 521     Must(makingVb == opUndecided);
 522     BodyPipePointer &p = theVirginRep.raw().body_pipe;
 523     Must(p != NULL);
 524     Must(p->setConsumerIfNotLate(this)); // to deliver vb, we must receive vb
 525     makingVb = opOn;
 526 }
 527
 528 void
 529 Adaptation::Ecap::XactionRep::vbStopMaking()
 530 {
 531     Must(makingVb == opOn);
 532     // if adapter does not need vb, we do not need to receive it
 533     sinkVb("vbStopMaking");
 534     Must(makingVb == opComplete);
 535 }
 536
 537 void
 538 Adaptation::Ecap::XactionRep::vbMakeMore()
 539 {
 540     Must(makingVb == opOn); // cannot make more if done proxying
 541     // we cannot guarantee more vb, but we can check that there is a chance
 542     const BodyPipePointer &p = theVirginRep.raw().body_pipe;
 543     Must(p != NULL && p->stillConsuming(this)); // we are plugged in
 544     Must(!p->productionEnded() && p->mayNeedMoreData()); // and may get more
 545 }
 546
 547 libecap::Area
 548 Adaptation::Ecap::XactionRep::vbContent(libecap::size_type o, libecap::size_type s)
 549 {
 550     // We may not be makingVb yet. It should be OK, but see vbContentShift().
 551
 552     const BodyPipePointer &p = theVirginRep.raw().body_pipe;
 553     Must(p != NULL);
 554
 555     // TODO: make MemBuf use size_t?
 556     const size_t haveSize = static_cast<size_t>(p->buf().contentSize());
 557
 558     // convert to Squid types; XXX: check for overflow
 559     const uint64_t offset = static_cast<uint64_t>(o);
 560     Must(offset <= haveSize); // equal iff at the end of content
 561
 562     // nsize means no size limit: all content starting from offset
 563     const size_t size = s == libecap::nsize ?
 564                         haveSize - offset : static_cast<size_t>(s);
 565
 566     // XXX: optimize by making theBody a shared_ptr (see Area::FromTemp*() src)
 567     return libecap::Area::FromTempBuffer(p->buf().content() + offset,
 568                                          min(static_cast<size_t>(haveSize - offset), size));
 569 }
 570
 571 void
 572 Adaptation::Ecap::XactionRep::vbContentShift(libecap::size_type n)
 573 {
 574     // We may not be makingVb yet. It should be OK now, but if BodyPipe
 575     // consume() requirements change, we would have to return empty vbContent
 576     // until the adapter registers as a consumer
 577
 578     BodyPipePointer &p = theVirginRep.raw().body_pipe;
 579     Must(p != NULL);
 580     const size_t size = static_cast<size_t>(n); // XXX: check for overflow
 581     const size_t haveSize = static_cast<size_t>(p->buf().contentSize()); // TODO: make MemBuf use size_t?
 582     p->consume(min(size, haveSize));
 583 }
 584
 585 void
 586 Adaptation::Ecap::XactionRep::noteAbContentDone(bool atEnd)
 587 {
 588     Must(proxyingAb == opOn && !abProductionFinished);
 589     abProductionFinished = true;
 590     abProductionAtEnd = atEnd; // store until ready to stop producing ourselves
 591     debugs(93,5, HERE << "adapted body production ended");
 592     moveAbContent();
 593 }
 594
 595 void
 596 Adaptation::Ecap::XactionRep::noteAbContentAvailable()
 597 {
 598     Must(proxyingAb == opOn && !abProductionFinished);
 599     moveAbContent();
 600 }
 601
 602 #if 0 /* XXX: implement */
 603 void
 604 Adaptation::Ecap::XactionRep::setAdaptedBodySize(const libecap::BodySize &size)
 605 {
 606     Must(answer().body_pipe != NULL);
 607     if (size.known())
 608         answer().body_pipe->setBodySize(size.value());
 609     // else the piped body size is unknown by default
 610 }
 611 #endif
 612
 613 void
 614 Adaptation::Ecap::XactionRep::adaptationDelayed(const libecap::Delay &d)
 615 {
 616     debugs(93,3, HERE << "adapter needs time: " <<
 617            d.state << '/' << d.progress);
 618     // XXX: set timeout?
 619 }
 620
 621 void
 622 Adaptation::Ecap::XactionRep::adaptationAborted()
 623 {
 624     tellQueryAborted(true); // should eCAP support retries?
 625     mustStop("adaptationAborted");
 626 }
 627
 628 void
 629 Adaptation::Ecap::XactionRep::noteMoreBodySpaceAvailable(RefCount<BodyPipe> bp)
 630 {
 631     Must(proxyingAb == opOn);
 632     moveAbContent();
 633 }
 634
 635 void
 636 Adaptation::Ecap::XactionRep::noteBodyConsumerAborted(RefCount<BodyPipe> bp)
 637 {
 638     Must(proxyingAb == opOn);
 639     stopProducingFor(answer().body_pipe, false);
 640     Must(theMaster);
 641     theMaster->abStopMaking();
 642     proxyingAb = opComplete;
 643 }
 644
 645 void
 646 Adaptation::Ecap::XactionRep::noteMoreBodyDataAvailable(RefCount<BodyPipe> bp)
 647 {
 648     Must(makingVb == opOn); // or we would not be registered as a consumer
 649     Must(theMaster);
 650     theMaster->noteVbContentAvailable();
 651 }
 652
 653 void
 654 Adaptation::Ecap::XactionRep::noteBodyProductionEnded(RefCount<BodyPipe> bp)
 655 {
 656     Must(makingVb == opOn); // or we would not be registered as a consumer
 657     Must(theMaster);
 658     theMaster->noteVbContentDone(true);
 659     vbProductionFinished = true;
 660 }
 661
 662 void
 663 Adaptation::Ecap::XactionRep::noteBodyProducerAborted(RefCount<BodyPipe> bp)
 664 {
 665     Must(makingVb == opOn); // or we would not be registered as a consumer
 666     Must(theMaster);
 667     theMaster->noteVbContentDone(false);
 668     vbProductionFinished = true;
 669 }
 670
 671 void
 672 Adaptation::Ecap::XactionRep::noteInitiatorAborted()
 673 {
 674     mustStop("initiator aborted");
 675 }
 676
 677 // get content from the adapter and put it into the adapted pipe
 678 void
 679 Adaptation::Ecap::XactionRep::moveAbContent()
 680 {
 681     Must(proxyingAb == opOn);
 682     const libecap::Area c = theMaster->abContent(0, libecap::nsize);
 683     debugs(93,5, HERE << "up to " << c.size << " bytes");
 684     if (c.size == 0 && abProductionFinished) { // no ab now and in the future
 685         stopProducingFor(answer().body_pipe, abProductionAtEnd);
 686         proxyingAb = opComplete;
 687         debugs(93,5, HERE << "last adapted body data retrieved");
 688     } else if (c.size > 0) {
 689         if (const size_t used = answer().body_pipe->putMoreData(c.start, c.size))
 690             theMaster->abContentShift(used);
 691     }
 692 }
 693
 694 const char *
 695 Adaptation::Ecap::XactionRep::status() const
 696 {
 697     static MemBuf buf;
 698     buf.reset();
 699
 700     buf.append(" [", 2);
 701
 702     if (makingVb)
 703         buf.appendf("M%d", static_cast<int>(makingVb));
 704
 705     const BodyPipePointer &vp = theVirginRep.raw().body_pipe;
 706     if (!vp)
 707         buf.append(" !V", 3);
 708     else if (vp->stillConsuming(const_cast<XactionRep*>(this)))
 709         buf.append(" Vc", 3);
 710     else
 711         buf.append(" V?", 3);
 712
 713     if (vbProductionFinished)
 714         buf.append(".", 1);
 715
 716     buf.appendf(" A%d", static_cast<int>(proxyingAb));
 717
 718     if (proxyingAb == opOn) {
 719         MessageRep *rep = dynamic_cast<MessageRep*>(theAnswerRep.get());
 720         Must(rep);
 721         const BodyPipePointer &ap = rep->raw().body_pipe;
 722         if (!ap)
 723             buf.append(" !A", 3);
 724         else if (ap->stillProducing(const_cast<XactionRep*>(this)))
 725             buf.append(" Ap", 3);
 726         else
 727             buf.append(" A?", 3);
 728     }
 729
 730     buf.appendf(" %s%u]", id.prefix(), id.value);
 731
 732     buf.terminate();
 733
 734     return buf.content();
 735 }
 736
 737 void
 738 Adaptation::Ecap::XactionRep::updateSources(Http::Message *adapted)
 739 {
 740     adapted->sources |= service().cfg().connectionEncryption ? Http::Message::srcEcaps : Http::Message::srcEcap;
 741
 742     // Update masterXaction object for adapted HTTP requests.
 743     if (HttpRequest *adaptedReq = dynamic_cast<HttpRequest*>(adapted)) {
 744         HttpRequest *request = dynamic_cast<HttpRequest*> (theCauseRep ?
 745                                theCauseRep->raw().header : theVirginRep.raw().header);
 746         Must(request);
 747         adaptedReq->masterXaction = request->masterXaction;
 748     }
 749 }
 750