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

// -*- mode: cpp; mode: fold -*-
// Description								/*{{{*/
// $Id: acquire-worker.cc,v 1.31 2000/05/10 05:56:46 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							/*{{{*/
#ifdef __GNUG__
#pragma implementation "apt-pkg/acquire-worker.h"
#endif
#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 <sys/stat.h>
#include <unistd.h>
#include <fcntl.h>
#include <signal.h>
#include <stdio.h>
#include <errno.h>
									/*}}}*/

// 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();

   // Spawn the subprocess
   if (Process == 0)
   {
      // Setup the FDs
      dup2(Pipes[1],STDOUT_FILENO);
      dup2(Pipes[2],STDIN_FILENO);
      dup2(((filebuf *)clog.rdbuf())->fd(),STDERR_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("Bizzar Error - File size is not what the server reported %s %u",
			       LookupTag(Message,"Size","0").c_str(),TotalSize);

	    Owner->Done(Message,atoi(LookupTag(Message,"Size","0").c_str()),
			LookupTag(Message,"MD5-Hash"),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);
	    Owner->Failed(Message,Config);
	    ItemDone();

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

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