/*
 * lftp - file transfer program
 *
 * Copyright (c) 1996-2013 by Alexander V. Lukyanov (lav@yars.free.net)
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

/* Usage notes:
 *
 * Call PreFilter() to add a filter to the beginning of the chain; these
 * filters are initialized only once for all data.  For example,
 * PreFilter("wc -l")
 *
 */

/*
 * Implementation notes:
 *
 * Background things we can't get around:
 * We must buffer (via FileCopy) output to a filter, since it might block.
 *
 * We must buffer the output from the filter to an output FileCopyPeer (ie.
 * a URL), for the same reason.
 *
 * So, we're stuck with having up to two FileCopy's.  (One to send, one to filter.)
 *
 * In some cases, we only need one: if the output is an FD, the filter can
 * hook up directly and we can forget about that stage.
 *
 * In the case where we're outputting to a URL, we set up a FileCopy from a
 * pipe to the URL, and then pretend we're just outputting to an FD (the
 * pipe.) This means in the simple case of having no filters at all, writing
 * to a URL or file, we send the data an extra time through a FileCopy and a
 * pipe.  That's a bit inefficient, but that's "cat file1 > file2"; that's
 * normally done with "get file1 -o file2", so this shouldn't happen often.
 *
 * It's very important that if the output is stdout, any filters point directly
 * at it, not through an extra copy: a pager, for example, will expect the output
 * to be a TTY.
 *
 */
#include <config.h>

#include "OutputJob.h"
#include "ArgV.h"
#include "FileCopy.h"
#include "CopyJob.h"
#include "url.h"
#include "misc.h"
#include "StatusLine.h"
#include "DummyProto.h"

#include <assert.h>
#include <unistd.h>
#include <errno.h>
#include <sys/ioctl.h>
#include <fcntl.h>
#include <stddef.h>

#define super Job

void OutputJob::InitCopy()
{
   if(error)
      return;

   if(initialized)
      return;

   if(fa)
   {
      /* Set up a pipe sending data at the peer, so we can act like the FDStream
       * constructor. */
      int filter_pipe[2];

      if(pipe(filter_pipe) == -1) {
	 // retry later
	 current->TimeoutS(1);
	 return;
      }

      FileCopyPeerFA *dst_peer = FileCopyPeerFA::New(fa.borrow(), fa_path, FA::STORE);

      /* Status only for remote outputs. */
      if(!strcmp(dst_peer->GetProto(), "file"))
	 no_status=true;

      fcntl(filter_pipe[0],F_SETFL,O_NONBLOCK);
      fcntl(filter_pipe[1],F_SETFL,O_NONBLOCK);

      /* The output of the pipe (0) goes to the output FileCopy. */
      FDStream *pipe_output = new FDStream(filter_pipe[0],"<filter-out>");

      FileCopy *output_fc=FileCopy::New(new FileCopyPeerFDStream(pipe_output, FileCopyPeer::GET), dst_peer,false);
      output=new CopyJob(output_fc, fa_path, a0);
      output->NoStatus(no_status);
      output_fd=new FDStream(filter_pipe[1],"<filter-in>");

      pipe_output->CloseWhenDone();
      output_fd->CloseWhenDone();

      fa_path.set(0);
   }

   initialized=true;

   if(Error())
      return;

   /* Clear the statusline, since we might change the pgrp if we create filters. */
   ClearStatus();

   /* Some legitimate uses produce broken pipe condition (cat|head).
    * We still want to produce broken pipe if we're not piping, eg
    * cat > pipe. */
   if(IsFiltered())
      fail_if_broken=false;

   if(filter)
   {
      /* Create the global filter: */
      OutputFilter *global = new OutputFilter(filter, output_fd.borrow());
      output_fd=global;
   }

   /* Use a FileCopy to buffer our output to the filter: */
   FileCopyPeerFDStream *out = new FileCopyPeerFDStream(output_fd.borrow(), FileCopyPeer::PUT);
   FileCopy *input_fc = FileCopy::New(new FileCopyPeer(FileCopyPeer::GET), out, false);

   if(!fail_if_broken)
      input_fc->DontFailIfBroken();

   input=new CopyJob(input_fc, xstring::format(_("%s (filter)"),a0.get()), filter?filter:a0);

   if(!output)
      output=input;

   input->SetParentFg(this);
   InputPeer()->SetDate(NO_DATE);
   InputPeer()->SetSize(NO_SIZE);
   input->GetCopy()->DontCopyDate();
   input->NoStatus();

   if(input != output)
   {
      output->SetParentFg(this);
      OutputPeer()->SetDate(NO_DATE);
      OutputPeer()->SetSize(NO_SIZE);
      output->GetCopy()->DontCopyDate();
      output->NoStatus();
   }

   if(is_stdout)
   {
      output->ClearStatusOnWrite();
      output->GetCopy()->LineBuffered();
   }

   Timeout(0);
}

void OutputJob::Init(const char *_a0)
{
   input=output=0;
   initialized=false;
   error=false;
   no_status=false;
   a0.set(_a0);
   is_stdout=false;
   fail_if_broken=true;
   is_a_tty=false;
   width=-1;
   statusbar_redisplay=true;
}

/* Local (fd) output. */
OutputJob::OutputJob(FDStream *output_, const char *a0)
   : output_fd(output_ ? output_ : new FDStream(1,"<stdout>"))
{
   Init(a0);

   if(output_)
   {
      /* We don't want to produce broken pipe when we're actually
       * piping, since some legitimate uses produce broken pipe, eg
       * cat|head.  However, that's actually handled in InitCopy().
       * User pipes aren't handled by us yet: instead of being set with
       * SetFilter, they're being set up ahead of time and passed to
       * us as an FDStream, so we don't really know if we're being filtered.
       *
       * So, until we handle pipes directly, disable broken pipe whenever
       * we're being sent anywhere but stdout. */
      fail_if_broken=false;
   }

   is_stdout=output_fd->usesfd(1);
   is_a_tty=isatty(output_fd->fd);
   width=fd_width(output_fd->fd);

   /* We don't output status when outputting locally. */
   no_status=true;

   /* Make sure that if the output is going to fail, it fails early, so
    * the parent doesn't start anything expensive (like begin downloading
    * a file.) */
   if(output_fd->getfd()==-1 && output_fd->error())
   {
      eprintf("%s: %s\n", a0, output_fd->error_text.get());
      error=true;
   }
}

OutputJob::OutputJob(const char *path, const char *a0, FileAccess *fa0)
   : fa(fa0 ? fa0->Clone() : FileAccess::New("file")), fa_path(path)
{
   Init(a0);
}

void OutputJob::PrepareToDie()
{
   Bg();
   AcceptSig(SIGTERM);

   Delete(input);
   if(input != output)
      Delete(output);
   input=0;
   output=0;

   super::PrepareToDie();
}

/* This is called to ask us "permission" to display a status line. */
bool OutputJob::ShowStatusLine(const SMTaskRef<StatusLine>& s)
{
   /* If our output file is gone, or isn't stdout, we don't care. */
   if(!output || !is_stdout)
      return true;

   /* If we're filtered, we never display at all.  (We don't know anything about
    * the filter's output; the only time we do is when we're outputting to a URL
    * or a file, and that doesn't apply here.) */
   if(IsFiltered())
      return false;

   /* If we're not line buffered, display only if the output CopyJob says to. */
   if(!output->GetCopy()->IsLineBuffered())
      return output->HasStatus();

   /* We're line buffered, so we can output a status line without stomping
    * on a partially output line.
    *
    * If we've output something recently, only send the output to the title,
    * to avoid flickering status for no reason.
    */
   if(!update_timer.Stopped()) {
      s->NextUpdateTitleOnly();
      return true;
   }

   /* If we're not reenabling the status bar, and the statusbar has
    * been turned off (due to output being reenabled), only send to
    * the title. */
   if(!statusbar_redisplay && output->GetCopy()->WriteAllowed())
   {
      s->NextUpdateTitleOnly();
      return true;
   }

   /* Don't disable write if there are data to be written in buffer */
   if(output->GetCopy()->WriteAllowed() && output->GetCopy()->WritePending())
      return false;

   /* There hasn't been output in a while.  Stop the output again,
    * so the FileCopy will clear the StatusLine when there's more data. */
   output->GetCopy()->AllowWrite(false);

   return true;
}

/* Get our contribution to the status line, which is just the output
 * status, if any.  Input status is the job of the user object. */
const char *OutputJob::Status(const StatusLine *s)
{
   if(no_status)
      return "";

   /* Never show anything if we havn't even received any data yet; it won't
    * start connecting until then, so it's not interesting. */
   if(!initialized)
      return "";

   /* Use the status from the output CopyJob.  It'll be the one that's connecting
    * to a host, if applicable. */
   return output->Status(s,true);
}

void OutputJob::PutEOF()
{
   if(Error())
      return;

   /* Make sure we've sent at least one (empty) block.  This ensures
    * that we always start the input->output code path. */
   Put("", 0);

   /* Send an EOF to the input peer; it'll send an EOF to the output peer
    * when all of its data is actually sent. */
   if(InputPeer())
      InputPeer()->PutEOF();
   else if(tmp_buf)
      tmp_buf->PutEOF();
   else
      abort();
}

/* add a filter to the beginning of the list */
void OutputJob::PreFilter(const char *newfilter)
{
   if(!filter)
      filter.set(newfilter);
   else
   {
      char *old_filter=alloca_strdup(filter);
      filter.vset(newfilter," | ",old_filter,NULL);
   }
}

/* Return the width of the output.  If there's a filter, we can either
 * return -1 (we might be piping through "sed", changing the width),
 * or the width we know (which is sane for most pagers.)  I'm not sure
 * which is better. */
int OutputJob::GetWidth() const
{
   if(IsFiltered())
      return -1;
   return width;
}

/* Return true if the output is going directly to a TTY. */
bool OutputJob::IsTTY() const
{
   if(IsFiltered())
      return false;
   return is_a_tty;
}

static const SMTaskRef<FileCopyPeer> null_peer;

/* Get the input FileCopyPeer; this is the buffer we write to. */
const SMTaskRef<FileCopyPeer>& OutputJob::InputPeer() const
{
   if(input)
      return input->GetGet();
   return null_peer;
}

/* Get the output FileCopyPeer (the FileCopyPeer that's doing the final output). */
const SMTaskRef<FileCopyPeer>& OutputJob::OutputPeer() const
{
   if(output)
      return output->GetPut();
   return null_peer;
}

/* We're done if the output is finished, or on error. */
int OutputJob::Done()
{
   if(Error())
      return true;

   if(!initialized)
      return false;

   if(input && !input->Done())
     return false;
   if(output && !output->Done())
     return false;

   return true;
}

int OutputJob::Do()
{
   if(!initialized && tmp_buf)
      InitCopy();
   return STALL;
}

/* Don't register errors until they're actually printed by
 * the sub-job (ie. it's also Done()). */
bool OutputJob::Error()
{
   if(error)
      return true;
   if(input && input->Error() && input->Done())
      error=true;
   if(output && input != output && output->Error() && output->Done())
      error=true;
   return error;
}

void OutputJob::Fg()
{
   super::Fg();
   if(input)
      input->Fg();
   if(output && input != output)
      output->Fg();
}

void OutputJob::Bg()
{
   if(output && input != output)
      output->Bg();
   if(input)
      input->Bg();
   super::Bg();
}

void OutputJob::SuspendInternal()
{
   super::SuspendInternal();
   if(input)
      input->SuspendSlave();
   if(output && input != output)
      output->SuspendSlave();
}

void OutputJob::ResumeInternal()
{
   if(input)
      input->ResumeSlave();
   if(output && input != output)
      output->ResumeSlave();
   super::ResumeInternal();
}

bool OutputJob::Full()
{
   /* It'd be nicer to just check copy->GetGet()->IsSuspended(), since
    * the FileCopy will suspend the Get end if the Put end gets filled.
    * However, it won't do that until it actually tries to send something. */
   int size = 0;
   if(input)
   {
      if(input->GetPut())
	 size += input->GetPut()->Buffered();
      if(input->GetGet())
	 size += input->GetGet()->Buffered();
      if(input != output)
      {
	 if(output->GetPut())
	    size += output->GetPut()->Buffered();
	 if(output->GetGet())
	    size += output->GetGet()->Buffered();
      }
   }
   if(tmp_buf)
      size += tmp_buf->Size();

   return size >= 0x10000;
}

/* We'll actually go over the buffer limit here; that's OK; it's not a
 * strict value.  (It's not convenient to prevent that completely with
 * Format(), either.) */
void OutputJob::Put(const char *buf,int size)
{
   InitCopy();
   if(Error())
      return;
   if(!InputPeer())
   {
      if(!tmp_buf)
	 tmp_buf=new Buffer;
      tmp_buf->Put(buf,size);
      return;
   }

   // InputPeer was inited, flush tmp_buf.
   if(InputPeer() && tmp_buf)
   {
      Ref<Buffer> saved_buf(tmp_buf.borrow());
      const char *b=0;
      int s=0;
      saved_buf->Get(&b,&s);
      if(b && s>0)
	 Put(b,s);
      if(saved_buf->Eof())
	 PutEOF();
   }

   update_timer.SetResource("cmd:status-interval",0);

   off_t oldpos = InputPeer()->GetPos();
   InputPeer()->Put(buf, size);
   InputPeer()->SetPos(oldpos);
}

void OutputJob::Format(const char *f,...)
{
   va_list v;
   va_start(v,f);
   const xstring& str=xstring::vformat(f,v);
   va_end(v);
   Put(str);
}

/* Propagate signals down to our child processes. */
int OutputJob::AcceptSig(int sig)
{
   int m=MOVED;
   if(sig == SIGTERM || sig == SIGINT)
      m=WANTDIE;

   /* If we have an input copier right now, it'll contain the top filter
    * (which is linked to all other filters), so send it the signal. */
   if(input)
      input->AcceptSig(sig);
   /* Otherwise, the only filters we have running are in output_fd. */
   else if(output_fd)
      output_fd->Kill(sig);
   if(sig!=SIGCONT)
      AcceptSig(SIGCONT);
   return m;
}