[apt.git] / apt-pkg / acquire-worker.cc

// -*- mode: cpp; mode: fold -*-
// Description								/*{{{*/
// $Id: acquire-worker.cc,v 1.34 2001/05/22 04:42:54 jgg Exp $
/* ######################################################################

   Acquire Worker 

   The worker process can startup either as a Configuration prober
   or as a queue runner. As a configuration prober it only reads the
   configuration message and 
   
   ##################################################################### */
									/*}}}*/
// Include Files							/*{{{*/
#include <apt-pkg/acquire-worker.h>
#include <apt-pkg/acquire-item.h>
#include <apt-pkg/configuration.h>
#include <apt-pkg/error.h>
#include <apt-pkg/fileutl.h>
#include <apt-pkg/strutl.h>

#include <apti18n.h>

#include <iostream>
#include <sstream>
#include <fstream>
    
#include <sys/stat.h>
#include <unistd.h>
#include <fcntl.h>
#include <signal.h>
#include <stdio.h>
#include <errno.h>
									/*}}}*/

using namespace std;

// Worker::Worker - Constructor for Queue startup			/*{{{*/
// ---------------------------------------------------------------------
/* */
pkgAcquire::Worker::Worker(Queue *Q,MethodConfig *Cnf,
			   pkgAcquireStatus *Log) : Log(Log)
{
   OwnerQ = Q;
   Config = Cnf;
   Access = Cnf->Access;
   CurrentItem = 0;
   TotalSize = 0;
   CurrentSize = 0;
   
   Construct();   
}
									/*}}}*/
// Worker::Worker - Constructor for method config startup		/*{{{*/
// ---------------------------------------------------------------------
/* */
pkgAcquire::Worker::Worker(MethodConfig *Cnf)
{
   OwnerQ = 0;
   Config = Cnf;
   Access = Cnf->Access;
   CurrentItem = 0;
   TotalSize = 0;
   CurrentSize = 0;
   
   Construct();   
}
									/*}}}*/
// Worker::Construct - Constructor helper				/*{{{*/
// ---------------------------------------------------------------------
/* */
void pkgAcquire::Worker::Construct()
{
   NextQueue = 0;
   NextAcquire = 0;
   Process = -1;
   InFd = -1;
   OutFd = -1;
   OutReady = false;
   InReady = false;
   Debug = _config->FindB("Debug::pkgAcquire::Worker",false);
}
									/*}}}*/
// Worker::~Worker - Destructor						/*{{{*/
// ---------------------------------------------------------------------
/* */
pkgAcquire::Worker::~Worker()
{
   close(InFd);
   close(OutFd);
   
   if (Process > 0)
   {
      /* Closing of stdin is the signal to exit and die when the process
         indicates it needs cleanup */
      if (Config->NeedsCleanup == false)
	 kill(Process,SIGINT);
      ExecWait(Process,Access.c_str(),true);
   }   
}
									/*}}}*/
// Worker::Start - Start the worker process				/*{{{*/
// ---------------------------------------------------------------------
/* This forks the method and inits the communication channel */
bool pkgAcquire::Worker::Start()
{
   // Get the method path
   string Method = _config->FindDir("Dir::Bin::Methods") + Access;
   if (FileExists(Method) == false)
      return _error->Error(_("The method driver %s could not be found."),Method.c_str());

   if (Debug == true)
      clog << "Starting method '" << Method << '\'' << endl;

   // Create the pipes
   int Pipes[4] = {-1,-1,-1,-1};
   if (pipe(Pipes) != 0 || pipe(Pipes+2) != 0)
   {
      _error->Errno("pipe","Failed to create IPC pipe to subprocess");
      for (int I = 0; I != 4; I++)
	 close(Pipes[I]);
      return false;
   }
   for (int I = 0; I != 4; I++)
      SetCloseExec(Pipes[I],true);
   
   // Fork off the process
   Process = ExecFork();
   if (Process == 0)
   {
      // Setup the FDs
      dup2(Pipes[1],STDOUT_FILENO);
      dup2(Pipes[2],STDIN_FILENO);
      SetCloseExec(STDOUT_FILENO,false);
      SetCloseExec(STDIN_FILENO,false);      
      SetCloseExec(STDERR_FILENO,false);
      
      const char *Args[2];
      Args[0] = Method.c_str();
      Args[1] = 0;
      execv(Args[0],(char **)Args);
      cerr << "Failed to exec method " << Args[0] << endl;
      _exit(100);
   }

   // Fix up our FDs
   InFd = Pipes[0];
   OutFd = Pipes[3];
   SetNonBlock(Pipes[0],true);
   SetNonBlock(Pipes[3],true);
   close(Pipes[1]);
   close(Pipes[2]);
   OutReady = false;
   InReady = true;
   
   // Read the configuration data
   if (WaitFd(InFd) == false ||
       ReadMessages() == false)
      return _error->Error(_("Method %s did not start correctly"),Method.c_str());

   RunMessages();
   if (OwnerQ != 0)
      SendConfiguration();
   
   return true;
}
									/*}}}*/
// Worker::ReadMessages - Read all pending messages into the list	/*{{{*/
// ---------------------------------------------------------------------
/* */
bool pkgAcquire::Worker::ReadMessages()
{
   if (::ReadMessages(InFd,MessageQueue) == false)
      return MethodFailure();
   return true;
}
									/*}}}*/
// Worker::RunMessage - Empty the message queue				/*{{{*/
// ---------------------------------------------------------------------
/* This takes the messages from the message queue and runs them through
   the parsers in order. */
bool pkgAcquire::Worker::RunMessages()
{
   while (MessageQueue.empty() == false)
   {
      string Message = MessageQueue.front();
      MessageQueue.erase(MessageQueue.begin());

      if (Debug == true)
	 clog << " <- " << Access << ':' << QuoteString(Message,"\n") << endl;
      
      // Fetch the message number
      char *End;
      int Number = strtol(Message.c_str(),&End,10);
      if (End == Message.c_str())
	 return _error->Error("Invalid message from method %s: %s",Access.c_str(),Message.c_str());

      string URI = LookupTag(Message,"URI");
      pkgAcquire::Queue::QItem *Itm = 0;
      if (URI.empty() == false)
	 Itm = OwnerQ->FindItem(URI,this);
      
      // Determine the message number and dispatch
      switch (Number)
      {
	 // 100 Capabilities
	 case 100:
	 if (Capabilities(Message) == false)
	    return _error->Error("Unable to process Capabilities message from %s",Access.c_str());
	 break;
	 
	 // 101 Log
	 case 101:
	 if (Debug == true)
	    clog << " <- (log) " << LookupTag(Message,"Message") << endl;
	 break;
	 
	 // 102 Status
	 case 102:
	 Status = LookupTag(Message,"Message");
	 break;
	    
	 // 200 URI Start
	 case 200:
	 {
	    if (Itm == 0)
	    {
	       _error->Error("Method gave invalid 200 URI Start message");
	       break;
	    }
	    
	    CurrentItem = Itm;
	    CurrentSize = 0;
	    TotalSize = atoi(LookupTag(Message,"Size","0").c_str());
	    ResumePoint = atoi(LookupTag(Message,"Resume-Point","0").c_str());
	    Itm->Owner->Start(Message,atoi(LookupTag(Message,"Size","0").c_str()));

	    // Display update before completion
	    if (Log != 0 && Log->MorePulses == true)
	       Log->Pulse(Itm->Owner->GetOwner());
	    
	    if (Log != 0)
	       Log->Fetch(*Itm);

	    break;
	 }
	 
	 // 201 URI Done
	 case 201:
	 {
	    if (Itm == 0)
	    {
	       _error->Error("Method gave invalid 201 URI Done message");
	       break;
	    }
	    
	    pkgAcquire::Item *Owner = Itm->Owner;
	    pkgAcquire::ItemDesc Desc = *Itm;
	    
	    // Display update before completion
	    if (Log != 0 && Log->MorePulses == true)
	       Log->Pulse(Owner->GetOwner());
	    
	    OwnerQ->ItemDone(Itm);
	    if (TotalSize != 0 &&
		(unsigned)atoi(LookupTag(Message,"Size","0").c_str()) != TotalSize)
	       _error->Warning("Bizarre Error - File size is not what the server reported %s %lu",
			       LookupTag(Message,"Size","0").c_str(),TotalSize);

	    // see if there is a hash to verify
	    string RecivedHash;
	    HashString expectedHash(Owner->HashSum());
	    if(!expectedHash.empty()) 
	    {
	       string hashTag = expectedHash.HashType()+"-Hash";
	       string hashSum = LookupTag(Message, hashTag.c_str());
	       if(!hashSum.empty())
		  RecivedHash = expectedHash.HashType() + ":" + hashSum;
	       if(_config->FindB("Debug::pkgAcquire::Auth", false) == true)
	       {
		  clog << "201 URI Done: " << Owner->DescURI() << endl
		       << "RecivedHash: " << RecivedHash << endl
		       << "ExpectedHash: " << expectedHash.toStr() 
		       << endl << endl;
	       }
	    }
	    Owner->Done(Message,atoi(LookupTag(Message,"Size","0").c_str()),
			RecivedHash.c_str(), Config);
	    ItemDone();
	    
	    // Log that we are done
	    if (Log != 0)
	    {
	       if (StringToBool(LookupTag(Message,"IMS-Hit"),false) == true ||
		   StringToBool(LookupTag(Message,"Alt-IMS-Hit"),false) == true)
	       {
		  /* Hide 'hits' for local only sources - we also manage to
		     hide gets */
		  if (Config->LocalOnly == false)
		     Log->IMSHit(Desc);
	       }	       
	       else
		  Log->Done(Desc);
	    }
	    break;
	 }	 
	 
	 // 400 URI Failure
	 case 400:
	 {
	    if (Itm == 0)
	    {
	       _error->Error("Method gave invalid 400 URI Failure message");
	       break;
	    }

	    // Display update before completion
	    if (Log != 0 && Log->MorePulses == true)
	       Log->Pulse(Itm->Owner->GetOwner());
	    
	    pkgAcquire::Item *Owner = Itm->Owner;
	    pkgAcquire::ItemDesc Desc = *Itm;
	    OwnerQ->ItemDone(Itm);

	    // set some status
	    if(LookupTag(Message,"FailReason") == "Timeout" || 
	       LookupTag(Message,"FailReason") == "TmpResolveFailure" ||
	       LookupTag(Message,"FailReason") == "ResolveFailure" ||
	       LookupTag(Message,"FailReason") == "ConnectionRefused") 
	       Owner->Status = pkgAcquire::Item::StatTransientNetworkError;

	    Owner->Failed(Message,Config);
	    ItemDone();

	    if (Log != 0)
	       Log->Fail(Desc);

	    break;
	 }	 
	 
	 // 401 General Failure
	 case 401:
	 _error->Error("Method %s General failure: %s",Access.c_str(),LookupTag(Message,"Message").c_str());
	 break;
	 
	 // 403 Media Change
	 case 403:
	 MediaChange(Message); 
	 break;
      }      
   }
   return true;
}
									/*}}}*/
// Worker::Capabilities - 100 Capabilities handler			/*{{{*/
// ---------------------------------------------------------------------
/* This parses the capabilities message and dumps it into the configuration
   structure. */
bool pkgAcquire::Worker::Capabilities(string Message)
{
   if (Config == 0)
      return true;
   
   Config->Version = LookupTag(Message,"Version");
   Config->SingleInstance = StringToBool(LookupTag(Message,"Single-Instance"),false);
   Config->Pipeline = StringToBool(LookupTag(Message,"Pipeline"),false);
   Config->SendConfig = StringToBool(LookupTag(Message,"Send-Config"),false);
   Config->LocalOnly = StringToBool(LookupTag(Message,"Local-Only"),false);
   Config->NeedsCleanup = StringToBool(LookupTag(Message,"Needs-Cleanup"),false);
   Config->Removable = StringToBool(LookupTag(Message,"Removable"),false);

   // Some debug text
   if (Debug == true)
   {
      clog << "Configured access method " << Config->Access << endl;
      clog << "Version:" << Config->Version <<
	      " SingleInstance:" << Config->SingleInstance <<
	      " Pipeline:" << Config->Pipeline << 
	      " SendConfig:" << Config->SendConfig << 
	      " LocalOnly: " << Config->LocalOnly << 
	      " NeedsCleanup: " << Config->NeedsCleanup << 
	      " Removable: " << Config->Removable << endl;
   }
   
   return true;
}
									/*}}}*/
// Worker::MediaChange - Request a media change				/*{{{*/
// ---------------------------------------------------------------------
/* */
bool pkgAcquire::Worker::MediaChange(string Message)
{
   int status_fd = _config->FindI("APT::Status-Fd",-1);
   if(status_fd > 0) 
   {
      string Media = LookupTag(Message,"Media");
      string Drive = LookupTag(Message,"Drive"); 
      ostringstream msg,status;
      ioprintf(msg,_("Please insert the disc labeled: "
		     "'%s' "
		     "in the drive '%s' and press enter."),
	       Media.c_str(),Drive.c_str());
      status << "media-change: "  // message
	     << Media  << ":"     // media
	     << Drive  << ":"     // drive
	     << msg.str()         // l10n message
	     << endl;
      write(status_fd, status.str().c_str(), status.str().size());
   }

   if (Log == 0 || Log->MediaChange(LookupTag(Message,"Media"),
				    LookupTag(Message,"Drive")) == false)
   {
      char S[300];
      snprintf(S,sizeof(S),"603 Media Changed\nFailed: true\n\n");
      if (Debug == true)
	 clog << " -> " << Access << ':' << QuoteString(S,"\n") << endl;
      OutQueue += S;
      OutReady = true;
      return true;
   }

   char S[300];
   snprintf(S,sizeof(S),"603 Media Changed\n\n");
   if (Debug == true)
      clog << " -> " << Access << ':' << QuoteString(S,"\n") << endl;
   OutQueue += S;
   OutReady = true;
   return true;
}
									/*}}}*/
// Worker::SendConfiguration - Send the config to the method		/*{{{*/
// ---------------------------------------------------------------------
/* */
bool pkgAcquire::Worker::SendConfiguration()
{
   if (Config->SendConfig == false)
      return true;

   if (OutFd == -1)
      return false;
   
   string Message = "601 Configuration\n";
   Message.reserve(2000);

   /* Write out all of the configuration directives by walking the 
      configuration tree */
   const Configuration::Item *Top = _config->Tree(0);
   for (; Top != 0;)
   {
      if (Top->Value.empty() == false)
      {
	 string Line = "Config-Item: " + QuoteString(Top->FullTag(),"=\"\n") + "=";
	 Line += QuoteString(Top->Value,"\n") + '\n';
	 Message += Line;
      }
      
      if (Top->Child != 0)
      {
	 Top = Top->Child;
	 continue;
      }
      
      while (Top != 0 && Top->Next == 0)
	 Top = Top->Parent;
      if (Top != 0)
	 Top = Top->Next;
   }   
   Message += '\n';

   if (Debug == true)
      clog << " -> " << Access << ':' << QuoteString(Message,"\n") << endl;
   OutQueue += Message;
   OutReady = true; 
   
   return true;
}
									/*}}}*/
// Worker::QueueItem - Add an item to the outbound queue		/*{{{*/
// ---------------------------------------------------------------------
/* Send a URI Acquire message to the method */
bool pkgAcquire::Worker::QueueItem(pkgAcquire::Queue::QItem *Item)
{
   if (OutFd == -1)
      return false;
   
   string Message = "600 URI Acquire\n";
   Message.reserve(300);
   Message += "URI: " + Item->URI;
   Message += "\nFilename: " + Item->Owner->DestFile;
   Message += Item->Owner->Custom600Headers();
   Message += "\n\n";
   
   if (Debug == true)
      clog << " -> " << Access << ':' << QuoteString(Message,"\n") << endl;
   OutQueue += Message;
   OutReady = true;
   
   return true;
}
									/*}}}*/
// Worker::OutFdRead - Out bound FD is ready				/*{{{*/
// ---------------------------------------------------------------------
/* */
bool pkgAcquire::Worker::OutFdReady()
{
   int Res;
   do
   {
      Res = write(OutFd,OutQueue.c_str(),OutQueue.length());
   }
   while (Res < 0 && errno == EINTR);
   
   if (Res <= 0)
      return MethodFailure();

   // Hmm.. this should never happen.
   if (Res < 0)
      return true;
   
   OutQueue.erase(0,Res);
   if (OutQueue.empty() == true)
      OutReady = false;
   
   return true;
}
									/*}}}*/
// Worker::InFdRead - In bound FD is ready				/*{{{*/
// ---------------------------------------------------------------------
/* */
bool pkgAcquire::Worker::InFdReady()
{
   if (ReadMessages() == false)
      return false;
   RunMessages();
   return true;
}
									/*}}}*/
// Worker::MethodFailure - Called when the method fails			/*{{{*/
// ---------------------------------------------------------------------
/* This is called when the method is belived to have failed, probably because
   read returned -1. */
bool pkgAcquire::Worker::MethodFailure()
{
   _error->Error("Method %s has died unexpectedly!",Access.c_str());
   
   ExecWait(Process,Access.c_str(),true);
   Process = -1;
   close(InFd);
   close(OutFd);
   InFd = -1;
   OutFd = -1;
   OutReady = false;
   InReady = false;
   OutQueue = string();
   MessageQueue.erase(MessageQueue.begin(),MessageQueue.end());
   
   return false;
}
									/*}}}*/
// Worker::Pulse - Called periodically 					/*{{{*/
// ---------------------------------------------------------------------
/* */
void pkgAcquire::Worker::Pulse()
{
   if (CurrentItem == 0)
      return;
 
   struct stat Buf;
   if (stat(CurrentItem->Owner->DestFile.c_str(),&Buf) != 0)
      return;
   CurrentSize = Buf.st_size;
   
   // Hmm? Should not happen...
   if (CurrentSize > TotalSize && TotalSize != 0)
      TotalSize = CurrentSize;
}
									/*}}}*/
// Worker::ItemDone - Called when the current item is finished		/*{{{*/
// ---------------------------------------------------------------------
/* */
void pkgAcquire::Worker::ItemDone()
{
   CurrentItem = 0;
   CurrentSize = 0;
   TotalSize = 0;
   Status = string();
}
									/*}}}*/
Commit	Line	Data
0118833a AL	1	// -- mode: cpp; mode: fold --
0118833a AL	2	// Description /{{{/
24a0d63a	3	// $Id: acquire-worker.cc,v 1.34 2001/05/22 04:42:54 jgg Exp $
0118833a AL	4	/* ######################################################################
	5
	6	Acquire Worker
	7
3b5421b4 AL	8	The worker process can startup either as a Configuration prober
	9	or as a queue runner. As a configuration prober it only reads the
	10	configuration message and
	11
0118833a AL	12	##################################################################### */
	13	/}}}/
	14	// Include Files /{{{/
0118833a	15	#include <apt-pkg/acquire-worker.h>
0a8a80e5	16	#include <apt-pkg/acquire-item.h>
3b5421b4 AL	17	#include <apt-pkg/configuration.h>
	18	#include <apt-pkg/error.h>
	19	#include <apt-pkg/fileutl.h>
cdcc6d34	20	#include <apt-pkg/strutl.h>
3b5421b4	21
b2e465d6	22	#include <apti18n.h>
24a0d63a AL	23
24a0d63a AL	24	#include <iostream>
80a26ed1	25	#include <sstream>
24a0d63a	26	#include <fstream>
b2e465d6	27
8267fe24	28	#include <sys/stat.h>
3b5421b4	29	#include <unistd.h>
b0db36b1	30	#include <fcntl.h>
3b5421b4	31	#include <signal.h>
542ec555	32	#include <stdio.h>
b0db36b1	33	#include <errno.h>
3b5421b4 AL	34	/}}}/
3b5421b4 AL	35
24a0d63a AL	36	using namespace std;
24a0d63a AL	37
3b5421b4 AL	38	// Worker::Worker - Constructor for Queue startup /{{{/
	39	// ---------------------------------------------------------------------
	40	/* */
8267fe24 AL	41	pkgAcquire::Worker::Worker(Queue Q,MethodConfig Cnf,
8267fe24 AL	42	pkgAcquireStatus *Log) : Log(Log)
3b5421b4 AL	43	{
3b5421b4 AL	44	OwnerQ = Q;
0a8a80e5 AL	45	Config = Cnf;
	46	Access = Cnf->Access;
	47	CurrentItem = 0;
18ef0a78 AL	48	TotalSize = 0;
	49	CurrentSize = 0;
	50
3b5421b4 AL	51	Construct();
	52	}
	53	/}}}/
	54	// Worker::Worker - Constructor for method config startup /{{{/
	55	// ---------------------------------------------------------------------
	56	/* */
	57	pkgAcquire::Worker::Worker(MethodConfig *Cnf)
	58	{
	59	OwnerQ = 0;
	60	Config = Cnf;
	61	Access = Cnf->Access;
0a8a80e5	62	CurrentItem = 0;
18ef0a78 AL	63	TotalSize = 0;
18ef0a78 AL	64	CurrentSize = 0;
0a8a80e5	65
3b5421b4 AL	66	Construct();
	67	}
	68	/}}}/
	69	// Worker::Construct - Constructor helper /{{{/
	70	// ---------------------------------------------------------------------
	71	/* */
	72	void pkgAcquire::Worker::Construct()
	73	{
0a8a80e5 AL	74	NextQueue = 0;
0a8a80e5 AL	75	NextAcquire = 0;
3b5421b4 AL	76	Process = -1;
	77	InFd = -1;
	78	OutFd = -1;
0a8a80e5 AL	79	OutReady = false;
0a8a80e5 AL	80	InReady = false;
3b5421b4 AL	81	Debug = _config->FindB("Debug::pkgAcquire::Worker",false);
	82	}
	83	/}}}/
	84	// Worker::~Worker - Destructor /{{{/
	85	// ---------------------------------------------------------------------
	86	/* */
	87	pkgAcquire::Worker::~Worker()
	88	{
	89	close(InFd);
	90	close(OutFd);
	91
	92	if (Process > 0)
0a8a80e5	93	{
8e5fc8f5 AL	94	/* Closing of stdin is the signal to exit and die when the process
	95	indicates it needs cleanup */
	96	if (Config->NeedsCleanup == false)
	97	kill(Process,SIGINT);
ddc1d8d0	98	ExecWait(Process,Access.c_str(),true);
0a8a80e5	99	}
3b5421b4 AL	100	}
	101	/}}}/
	102	// Worker::Start - Start the worker process /{{{/
	103	// ---------------------------------------------------------------------
	104	/* This forks the method and inits the communication channel */
	105	bool pkgAcquire::Worker::Start()
	106	{
	107	// Get the method path
	108	string Method = _config->FindDir("Dir::Bin::Methods") + Access;
	109	if (FileExists(Method) == false)
b2e465d6	110	return _error->Error(_("The method driver %s could not be found."),Method.c_str());
3b5421b4 AL	111
	112	if (Debug == true)
	113	clog << "Starting method '" << Method << '\'' << endl;
	114
	115	// Create the pipes
	116	int Pipes[4] = {-1,-1,-1,-1};
	117	if (pipe(Pipes) != 0 \|\| pipe(Pipes+2) != 0)
	118	{
	119	_error->Errno("pipe","Failed to create IPC pipe to subprocess");
	120	for (int I = 0; I != 4; I++)
	121	close(Pipes[I]);
	122	return false;
	123	}
8b89e57f	124	for (int I = 0; I != 4; I++)
4490f2de	125	SetCloseExec(Pipes[I],true);
8b89e57f	126
3b5421b4	127	// Fork off the process
54676e1a	128	Process = ExecFork();
3b5421b4 AL	129	if (Process == 0)
	130	{
	131	// Setup the FDs
	132	dup2(Pipes[1],STDOUT_FILENO);
	133	dup2(Pipes[2],STDIN_FILENO);
3b5421b4 AL	134	SetCloseExec(STDOUT_FILENO,false);
	135	SetCloseExec(STDIN_FILENO,false);
	136	SetCloseExec(STDERR_FILENO,false);
	137
	138	const char *Args[2];
	139	Args[0] = Method.c_str();
	140	Args[1] = 0;
	141	execv(Args[0],(char **)Args);
	142	cerr << "Failed to exec method " << Args[0] << endl;
0dbb95d8	143	_exit(100);
3b5421b4 AL	144	}
	145
	146	// Fix up our FDs
	147	InFd = Pipes[0];
	148	OutFd = Pipes[3];
	149	SetNonBlock(Pipes[0],true);
	150	SetNonBlock(Pipes[3],true);
	151	close(Pipes[1]);
	152	close(Pipes[2]);
0a8a80e5 AL	153	OutReady = false;
0a8a80e5 AL	154	InReady = true;
3b5421b4 AL	155
	156	// Read the configuration data
	157	if (WaitFd(InFd) == false \|\|
	158	ReadMessages() == false)
b2e465d6	159	return _error->Error(_("Method %s did not start correctly"),Method.c_str());
3b5421b4 AL	160
3b5421b4 AL	161	RunMessages();
8b89e57f AL	162	if (OwnerQ != 0)
8b89e57f AL	163	SendConfiguration();
3b5421b4 AL	164
	165	return true;
	166	}
	167	/}}}/
	168	// Worker::ReadMessages - Read all pending messages into the list /{{{/
	169	// ---------------------------------------------------------------------
0a8a80e5	170	/* */
3b5421b4 AL	171	bool pkgAcquire::Worker::ReadMessages()
3b5421b4 AL	172	{
0a8a80e5 AL	173	if (::ReadMessages(InFd,MessageQueue) == false)
0a8a80e5 AL	174	return MethodFailure();
3b5421b4 AL	175	return true;
	176	}
	177	/}}}/
3b5421b4 AL	178	// Worker::RunMessage - Empty the message queue /{{{/
	179	// ---------------------------------------------------------------------
	180	/* This takes the messages from the message queue and runs them through
	181	the parsers in order. */
	182	bool pkgAcquire::Worker::RunMessages()
	183	{
	184	while (MessageQueue.empty() == false)
	185	{
	186	string Message = MessageQueue.front();
	187	MessageQueue.erase(MessageQueue.begin());
0a8a80e5 AL	188
	189	if (Debug == true)
	190	clog << " <- " << Access << ':' << QuoteString(Message,"\n") << endl;
3b5421b4 AL	191
	192	// Fetch the message number
	193	char *End;
	194	int Number = strtol(Message.c_str(),&End,10);
	195	if (End == Message.c_str())
	196	return _error->Error("Invalid message from method %s: %s",Access.c_str(),Message.c_str());
	197
c88edf1d AL	198	string URI = LookupTag(Message,"URI");
	199	pkgAcquire::Queue::QItem *Itm = 0;
	200	if (URI.empty() == false)
	201	Itm = OwnerQ->FindItem(URI,this);
bfd22fc0	202
3b5421b4 AL	203	// Determine the message number and dispatch
	204	switch (Number)
	205	{
0a8a80e5	206	// 100 Capabilities
3b5421b4 AL	207	case 100:
	208	if (Capabilities(Message) == false)
	209	return _error->Error("Unable to process Capabilities message from %s",Access.c_str());
	210	break;
0a8a80e5 AL	211
	212	// 101 Log
	213	case 101:
	214	if (Debug == true)
	215	clog << " <- (log) " << LookupTag(Message,"Message") << endl;
	216	break;
	217
	218	// 102 Status
	219	case 102:
	220	Status = LookupTag(Message,"Message");
	221	break;
	222
	223	// 200 URI Start
	224	case 200:
c88edf1d AL	225	{
	226	if (Itm == 0)
	227	{
93bf083d	228	_error->Error("Method gave invalid 200 URI Start message");
c88edf1d AL	229	break;
c88edf1d AL	230	}
8267fe24	231
c88edf1d AL	232	CurrentItem = Itm;
	233	CurrentSize = 0;
	234	TotalSize = atoi(LookupTag(Message,"Size","0").c_str());
8b75eb1c	235	ResumePoint = atoi(LookupTag(Message,"Resume-Point","0").c_str());
8267fe24	236	Itm->Owner->Start(Message,atoi(LookupTag(Message,"Size","0").c_str()));
8b75eb1c	237
c5ccf175 AL	238	// Display update before completion
	239	if (Log != 0 && Log->MorePulses == true)
	240	Log->Pulse(Itm->Owner->GetOwner());
	241
8267fe24 AL	242	if (Log != 0)
	243	Log->Fetch(*Itm);
	244
c88edf1d AL	245	break;
c88edf1d AL	246	}
0a8a80e5 AL	247
	248	// 201 URI Done
	249	case 201:
c88edf1d AL	250	{
	251	if (Itm == 0)
	252	{
93bf083d	253	_error->Error("Method gave invalid 201 URI Done message");
c88edf1d AL	254	break;
c88edf1d AL	255	}
c5ccf175	256
bfd22fc0	257	pkgAcquire::Item *Owner = Itm->Owner;
8267fe24	258	pkgAcquire::ItemDesc Desc = *Itm;
c5ccf175 AL	259
	260	// Display update before completion
	261	if (Log != 0 && Log->MorePulses == true)
	262	Log->Pulse(Owner->GetOwner());
	263
be4401bf	264	OwnerQ->ItemDone(Itm);
b2e465d6	265	if (TotalSize != 0 &&
bf3abeed	266	(unsigned)atoi(LookupTag(Message,"Size","0").c_str()) != TotalSize)
b2e465d6	267	_error->Warning("Bizarre Error - File size is not what the server reported %s %lu",
18ef0a78	268	LookupTag(Message,"Size","0").c_str(),TotalSize);
bdae53f1	269
8a8feb29 MV	270	// see if there is a hash to verify
	271	string RecivedHash;
	272	HashString expectedHash(Owner->HashSum());
	273	if(!expectedHash.empty())
	274	{
	275	string hashTag = expectedHash.HashType()+"-Hash";
95f45727 MV	276	string hashSum = LookupTag(Message, hashTag.c_str());
	277	if(!hashSum.empty())
	278	RecivedHash = expectedHash.HashType() + ":" + hashSum;
8a8feb29 MV	279	if(_config->FindB("Debug::pkgAcquire::Auth", false) == true)
	280	{
	281	clog << "201 URI Done: " << Owner->DescURI() << endl
	282	<< "RecivedHash: " << RecivedHash << endl
	283	<< "ExpectedHash: " << expectedHash.toStr()
	284	<< endl << endl;
	285	}
	286	}
bdae53f1	287	Owner->Done(Message,atoi(LookupTag(Message,"Size","0").c_str()),
8a8feb29	288	RecivedHash.c_str(), Config);
8267fe24 AL	289	ItemDone();
	290
	291	// Log that we are done
	292	if (Log != 0)
	293	{
	294	if (StringToBool(LookupTag(Message,"IMS-Hit"),false) == true \|\|
	295	StringToBool(LookupTag(Message,"Alt-IMS-Hit"),false) == true)
c46824ce AL	296	{
	297	/* Hide 'hits' for local only sources - we also manage to
	298	hide gets */
	299	if (Config->LocalOnly == false)
	300	Log->IMSHit(Desc);
	301	}
8267fe24 AL	302	else
8267fe24 AL	303	Log->Done(Desc);
18ef0a78	304	}
c88edf1d AL	305	break;
c88edf1d AL	306	}
0a8a80e5 AL	307
	308	// 400 URI Failure
	309	case 400:
c88edf1d AL	310	{
	311	if (Itm == 0)
	312	{
93bf083d	313	_error->Error("Method gave invalid 400 URI Failure message");
c88edf1d AL	314	break;
	315	}
	316
c5ccf175 AL	317	// Display update before completion
	318	if (Log != 0 && Log->MorePulses == true)
	319	Log->Pulse(Itm->Owner->GetOwner());
	320
bfd22fc0	321	pkgAcquire::Item *Owner = Itm->Owner;
8267fe24	322	pkgAcquire::ItemDesc Desc = *Itm;
c88edf1d	323	OwnerQ->ItemDone(Itm);
7e5f33eb MV	324
	325	// set some status
	326	if(LookupTag(Message,"FailReason") == "Timeout" \|\|
	327	LookupTag(Message,"FailReason") == "TmpResolveFailure" \|\|
f0983ff2	328	LookupTag(Message,"FailReason") == "ResolveFailure" \|\|
7e5f33eb MV	329	LookupTag(Message,"FailReason") == "ConnectionRefused")
	330	Owner->Status = pkgAcquire::Item::StatTransientNetworkError;
	331
7d8afa39	332	Owner->Failed(Message,Config);
8267fe24	333	ItemDone();
7d8afa39	334
8267fe24 AL	335	if (Log != 0)
8267fe24 AL	336	Log->Fail(Desc);
7d8afa39	337
c88edf1d AL	338	break;
c88edf1d AL	339	}
0a8a80e5 AL	340
	341	// 401 General Failure
	342	case 401:
b2e465d6	343	_error->Error("Method %s General failure: %s",Access.c_str(),LookupTag(Message,"Message").c_str());
0a8a80e5	344	break;
542ec555 AL	345
	346	// 403 Media Change
	347	case 403:
	348	MediaChange(Message);
	349	break;
3b5421b4 AL	350	}
	351	}
	352	return true;
	353	}
	354	/}}}/
	355	// Worker::Capabilities - 100 Capabilities handler /{{{/
	356	// ---------------------------------------------------------------------
	357	/* This parses the capabilities message and dumps it into the configuration
	358	structure. */
	359	bool pkgAcquire::Worker::Capabilities(string Message)
	360	{
	361	if (Config == 0)
	362	return true;
	363
	364	Config->Version = LookupTag(Message,"Version");
	365	Config->SingleInstance = StringToBool(LookupTag(Message,"Single-Instance"),false);
0a8a80e5 AL	366	Config->Pipeline = StringToBool(LookupTag(Message,"Pipeline"),false);
0a8a80e5 AL	367	Config->SendConfig = StringToBool(LookupTag(Message,"Send-Config"),false);
e331f6ed	368	Config->LocalOnly = StringToBool(LookupTag(Message,"Local-Only"),false);
8e5fc8f5	369	Config->NeedsCleanup = StringToBool(LookupTag(Message,"Needs-Cleanup"),false);
459681d3	370	Config->Removable = StringToBool(LookupTag(Message,"Removable"),false);
3b5421b4 AL	371
	372	// Some debug text
	373	if (Debug == true)
	374	{
	375	clog << "Configured access method " << Config->Access << endl;
459681d3 AL	376	clog << "Version:" << Config->Version <<
	377	" SingleInstance:" << Config->SingleInstance <<
	378	" Pipeline:" << Config->Pipeline <<
	379	" SendConfig:" << Config->SendConfig <<
	380	" LocalOnly: " << Config->LocalOnly <<
	381	" NeedsCleanup: " << Config->NeedsCleanup <<
	382	" Removable: " << Config->Removable << endl;
3b5421b4 AL	383	}
3b5421b4 AL	384
542ec555 AL	385	return true;
	386	}
	387	/}}}/
	388	// Worker::MediaChange - Request a media change /{{{/
	389	// ---------------------------------------------------------------------
	390	/* */
	391	bool pkgAcquire::Worker::MediaChange(string Message)
	392	{
80a26ed1 MV	393	int status_fd = _config->FindI("APT::Status-Fd",-1);
	394	if(status_fd > 0)
	395	{
	396	string Media = LookupTag(Message,"Media");
	397	string Drive = LookupTag(Message,"Drive");
	398	ostringstream msg,status;
1a82c63e MV	399	ioprintf(msg,_("Please insert the disc labeled: "
	400	"'%s' "
	401	"in the drive '%s' and press enter."),
	402	Media.c_str(),Drive.c_str());
80a26ed1	403	status << "media-change: " // message
1a82c63e MV	404	<< Media << ":" // media
	405	<< Drive << ":" // drive
	406	<< msg.str() // l10n message
80a26ed1 MV	407	<< endl;
	408	write(status_fd, status.str().c_str(), status.str().size());
	409	}
	410
542ec555 AL	411	if (Log == 0 \|\| Log->MediaChange(LookupTag(Message,"Media"),
	412	LookupTag(Message,"Drive")) == false)
	413	{
	414	char S[300];
96bc43c4	415	snprintf(S,sizeof(S),"603 Media Changed\nFailed: true\n\n");
542ec555 AL	416	if (Debug == true)
	417	clog << " -> " << Access << ':' << QuoteString(S,"\n") << endl;
	418	OutQueue += S;
	419	OutReady = true;
	420	return true;
	421	}
	422
	423	char S[300];
96bc43c4	424	snprintf(S,sizeof(S),"603 Media Changed\n\n");
542ec555 AL	425	if (Debug == true)
	426	clog << " -> " << Access << ':' << QuoteString(S,"\n") << endl;
	427	OutQueue += S;
	428	OutReady = true;
3b5421b4 AL	429	return true;
3b5421b4 AL	430	}
0118833a	431	/}}}/
0a8a80e5 AL	432	// Worker::SendConfiguration - Send the config to the method /{{{/
	433	// ---------------------------------------------------------------------
	434	/* */
	435	bool pkgAcquire::Worker::SendConfiguration()
	436	{
	437	if (Config->SendConfig == false)
	438	return true;
	439
	440	if (OutFd == -1)
	441	return false;
	442
	443	string Message = "601 Configuration\n";
	444	Message.reserve(2000);
	445
	446	/* Write out all of the configuration directives by walking the
	447	configuration tree */
	448	const Configuration::Item *Top = _config->Tree(0);
	449	for (; Top != 0;)
	450	{
	451	if (Top->Value.empty() == false)
	452	{
b2e465d6	453	string Line = "Config-Item: " + QuoteString(Top->FullTag(),"=\"\n") + "=";
0a8a80e5 AL	454	Line += QuoteString(Top->Value,"\n") + '\n';
	455	Message += Line;
	456	}
	457
	458	if (Top->Child != 0)
	459	{
	460	Top = Top->Child;
	461	continue;
	462	}
	463
	464	while (Top != 0 && Top->Next == 0)
	465	Top = Top->Parent;
	466	if (Top != 0)
	467	Top = Top->Next;
	468	}
	469	Message += '\n';
	470
	471	if (Debug == true)
	472	clog << " -> " << Access << ':' << QuoteString(Message,"\n") << endl;
	473	OutQueue += Message;
	474	OutReady = true;
	475
	476	return true;
	477	}
	478	/}}}/
	479	// Worker::QueueItem - Add an item to the outbound queue /{{{/
	480	// ---------------------------------------------------------------------
	481	/* Send a URI Acquire message to the method */
	482	bool pkgAcquire::Worker::QueueItem(pkgAcquire::Queue::QItem *Item)
	483	{
	484	if (OutFd == -1)
	485	return false;
	486
	487	string Message = "600 URI Acquire\n";
	488	Message.reserve(300);
	489	Message += "URI: " + Item->URI;
	490	Message += "\nFilename: " + Item->Owner->DestFile;
	491	Message += Item->Owner->Custom600Headers();
	492	Message += "\n\n";
	493
	494	if (Debug == true)
	495	clog << " -> " << Access << ':' << QuoteString(Message,"\n") << endl;
	496	OutQueue += Message;
	497	OutReady = true;
	498
	499	return true;
	500	}
	501	/}}}/
	502	// Worker::OutFdRead - Out bound FD is ready /{{{/
	503	// ---------------------------------------------------------------------
	504	/* */
	505	bool pkgAcquire::Worker::OutFdReady()
	506	{
b0db36b1 AL	507	int Res;
	508	do
	509	{
24a0d63a	510	Res = write(OutFd,OutQueue.c_str(),OutQueue.length());
b0db36b1 AL	511	}
	512	while (Res < 0 && errno == EINTR);
	513
0a8a80e5 AL	514	if (Res <= 0)
	515	return MethodFailure();
	516
	517	// Hmm.. this should never happen.
	518	if (Res < 0)
	519	return true;
	520
	521	OutQueue.erase(0,Res);
	522	if (OutQueue.empty() == true)
	523	OutReady = false;
	524
	525	return true;
	526	}
	527	/}}}/
	528	// Worker::InFdRead - In bound FD is ready /{{{/
	529	// ---------------------------------------------------------------------
	530	/* */
	531	bool pkgAcquire::Worker::InFdReady()
	532	{
	533	if (ReadMessages() == false)
	534	return false;
	535	RunMessages();
	536	return true;
	537	}
	538	/}}}/
	539	// Worker::MethodFailure - Called when the method fails /{{{/
	540	// ---------------------------------------------------------------------
	541	/* This is called when the method is belived to have failed, probably because
	542	read returned -1. */
	543	bool pkgAcquire::Worker::MethodFailure()
	544	{
76d97c26 AL	545	_error->Error("Method %s has died unexpectedly!",Access.c_str());
76d97c26 AL	546
ddc1d8d0	547	ExecWait(Process,Access.c_str(),true);
0a8a80e5 AL	548	Process = -1;
	549	close(InFd);
	550	close(OutFd);
	551	InFd = -1;
	552	OutFd = -1;
	553	OutReady = false;
	554	InReady = false;
	555	OutQueue = string();
	556	MessageQueue.erase(MessageQueue.begin(),MessageQueue.end());
	557
	558	return false;
	559	}
	560	/}}}/
8267fe24 AL	561	// Worker::Pulse - Called periodically /{{{/
	562	// ---------------------------------------------------------------------
	563	/* */
	564	void pkgAcquire::Worker::Pulse()
	565	{
	566	if (CurrentItem == 0)
	567	return;
542ec555	568
8267fe24 AL	569	struct stat Buf;
	570	if (stat(CurrentItem->Owner->DestFile.c_str(),&Buf) != 0)
	571	return;
	572	CurrentSize = Buf.st_size;
18ef0a78 AL	573
	574	// Hmm? Should not happen...
	575	if (CurrentSize > TotalSize && TotalSize != 0)
	576	TotalSize = CurrentSize;
8267fe24 AL	577	}
	578	/}}}/
	579	// Worker::ItemDone - Called when the current item is finished /{{{/
	580	// ---------------------------------------------------------------------
	581	/* */
	582	void pkgAcquire::Worker::ItemDone()
	583	{
	584	CurrentItem = 0;
	585	CurrentSize = 0;
	586	TotalSize = 0;
	587	Status = string();
	588	}
	589	/}}}/