Workspace.cpp #3

//
// Created by Tristan Juricek on 9/19/15.
//

#include <algorithm>
#include <iostream>
#include <map>
#include <memory>
#include <mutex>
#include <p4/clientapi.h>
#include <pwd.h>
#include <set>
#include <string>
#include <sstream>
#include <sys/types.h>
#include <unistd.h>
#include <uuid/uuid.h>
#include <vector>

#include "Timer.h"
#include "Workspace.h"

using std::back_inserter;
using std::chrono::system_clock;
using std::chrono::milliseconds;
using std::cout;
using std::cerr;
using std::endl;
using std::find;
using std::map;
using std::mutex;
using std::set;
using std::shared_ptr;
using std::string;
using std::stringstream;
using std::transform;
using std::vector;


class ServerMessage
{
public:
    string text;
    int severity;
    int code;

    ServerMessage() = default;

    ServerMessage(const Error *err)
    {
        StrBuf strBuf;
        err->Fmt(&strBuf);
        text = strBuf.Text();

        severity = err->GetSeverity();
        code = err->GetGeneric();
    }
};

//----------------------------------------------------------------------------
// CommandResult
//----------------------------------------------------------------------------

// Encapsulates both the return from a command, and the collected results.
//
class CommandResult
{
public:
    CommandResult(const vector<ServerMessage> & messages,
                  const vector<map<string, string>> & results)
    : _messages(messages), _results(results)
    { }

    ~CommandResult() = default;

    bool
    hasError() const
    {
        for (ServerMessage m : _messages)
        {
            if (m.severity >= E_FAILED)
                return true;
        }
        return false;
    }

    const vector<ServerMessage> &
    messages() const
    {
        return _messages;
    }

    const vector<map<string, string>> &
    results() const
    {
        return _results;
    }

    int processReturn() const
    {
        if (hasError())
        {
            for (ServerMessage message : _messages)
            {
                cerr << message.code << ": " << message.text << endl;
            }
            return Workspace::COMMAND_FAILURE;
        }
        return 0;
    }

    void log() const
    {
        if (!_messages.empty())
        {
            cout << "messages:" << endl;
            for (ServerMessage message : _messages)
            {
                cout << '\t' << message.code << ": " << message.text << endl;
            }
        }
        if (!_results.empty())
        {
            for (auto result : _results)
            {
                cout << "{" << endl;
                for (auto entry : result)
                {
                    cout << "  " << entry.first << " -> " << entry.second <<
                    endl;
                }
                cout << "}" << endl;
            }
        }
    }

private:
    vector<ServerMessage> _messages;
    vector<map<string, string>> _results;
};


//----------------------------------------------------------------------------
// CachingClientUser
//----------------------------------------------------------------------------

// Handles converting much of the output from the ClientApi to something
// less ... attached... to the p4api code.
class CachingClientUser : public ClientUser
{
public:
    CachingClientUser() = default;

    ~CachingClientUser() = default;

    virtual void OutputStat(StrDict *varList) override;

    virtual void Message(Error *err) override;

    const vector<map<string, string>> &
    results() const
    {
        return _results;
    };

    const vector<ServerMessage> &
    messages() const
    {
        return _messages;
    }

    void
    clear()
    {
        _results.clear();
        _messages.clear();
    }

private:
    vector<map<string, string>> _results;
    vector<ServerMessage> _messages;
};

void CachingClientUser::OutputStat(StrDict *varList)
{
    StrRef var, val;
    map<string, string> result;

    for (int index = 0; varList->GetVar(index, var, val); ++index)
    {
        if (var == "specdef" || var == "func" || var == "specFormatted")
            continue;

        result[var.Text()] = val.Text();
    }

    _results.push_back(result);
}

void CachingClientUser::Message(Error *err)
{
    _messages.push_back(ServerMessage(err));
}


//----------------------------------------------------------------------------
// ClientApiHandle
//----------------------------------------------------------------------------

// Use RAII around connecting and disconnecting to the Perforce server.
// We do not handle re-using connections just yet.
//
// Since the ClientApi is rather verbose, this contains some high-level methods
// to collate data.
class ClientApiHandle
{
public:
    static ClientApiHandle *create(string workspace);

    ClientApiHandle();

    ~ClientApiHandle();

    const ClientApi & api() const
    {
        return *_api;
    }

    ClientApi & api()
    {
        return *_api;
    }

    // High-level method to execute the command with optional args and return
    // results.
    template<typename... Strings>
    CommandResult
    run(const string & cmd, Strings...args);

    CommandResult
    run(const string & cmd, vector<const char *> argv);

    CommandResult
    run(const string & cmd, vector<string> argv);

    bool isConnected() const
    {
        return _connected;
    }

    void setConnected(bool c)
    {
        _connected = c;
    }

private:
    CachingClientUser *_user;
    ClientApi *_api;

    bool _connected;
};

// This will set the context of the ClientApi to the workspace path.
// It's assumed that the workspace path has a .p4config file set up with
// proper connection settings, etc, for you to commence work.
ClientApiHandle *
ClientApiHandle::create(string workspace)
{
    ClientApiHandle *handle = new ClientApiHandle;

    handle->api().SetCwd(workspace.c_str());

    // Setting the ticket file explicitly to the home directory, mostly because
    // it's our current default convention.
    //
    // TODO There has to be a better way to figure out the ticket file
    string ticketFile;
    struct passwd *pwdinfo = getpwuid(getuid());
    if (pwdinfo)
    {
        ticketFile = pwdinfo->pw_dir;
    }
    ticketFile = ticketFile + "/.p4tickets";
    cout << "Setting ticket file " << ticketFile << endl;

    handle->api().SetTicketFile(ticketFile.c_str());

    return handle;
}

ClientApiHandle::ClientApiHandle()
: _user(new CachingClientUser), _api(new ClientApi),
  _connected(false)
{
}

ClientApiHandle::~ClientApiHandle()
{
    if (_connected)
    {
        Error e;
        _api->Final(&e);

        if (e.IsError())
        {
            StrBuf strBuf;
            e.Fmt(&strBuf);
            cerr << "ClientApi::Final failed: " << strBuf.Text() << endl;
        }
    }

    delete _api;
    delete _user;
}

void
collect(vector<const char *> & v)
{
    // Do nothing
}

void
collect(vector<const char *> & v, const string & a)
{
    v.push_back(a.c_str());
}

template<typename... Strings>
void
collect(vector<const char *> & v, const string & a, Strings...rest)
{
    v.push_back(a.c_str());
    return collect(v, rest...);
}

template<typename... Strings>
CommandResult
ClientApiHandle::run(const string & cmd, Strings... args)
{
    vector<const char *> argv;
    collect(argv, args...);

    return run(cmd, argv);
}

CommandResult
ClientApiHandle::run(const string & cmd, vector<const char *> argv)
{
    _user->clear();

    if (!argv.empty())
    {
        _api->SetArgv(argv.size(), const_cast<char **>(argv.data()));
    }

    _api->Run(cmd.c_str(), _user);

    return CommandResult(_user->messages(), _user->results());
}

CommandResult
ClientApiHandle::run(const string & cmd, vector<string> argv)
{
    vector<const char *> chargv;
    transform(argv.begin(), argv.end(), back_inserter(chargv),
              [](string & s) -> const char * { return s.c_str(); });

    return run(cmd, chargv);
}

// Will call "p4 opened" and then pass those depot paths to "p4 where"
int loadLocalOpenedPaths(ClientApiHandle & handle, vector<string> & paths)
{
    CommandResult openedResult = handle.run("opened");
    if (openedResult.hasError())
        return openedResult.processReturn();

    // If there's no results, there's really nothing to do
    if (openedResult.results().empty())
        return 0;

    vector<string> depotPaths;
    transform(openedResult.results().begin(),
              openedResult.results().end(),
              back_inserter(depotPaths),
              [](const map<string, string> & m) -> string {
                  return m.at("depotFile");
              });

    CommandResult whereResult = handle.run("where", depotPaths);
    if (whereResult.hasError())
        return whereResult.processReturn();

    transform(whereResult.results().begin(),
              whereResult.results().end(),
              back_inserter(paths),
              [](const map<string, string> & m) -> string {
                  return m.at("path");
              });

    return 0;
}


//----------------------------------------------------------------------------
// WorkspaceImpl
//----------------------------------------------------------------------------

// We use a PIMPL technique on this class to ensure use of the P4API is
// restricted to *only* this file.
class WorkspaceImpl
{
public:
    WorkspaceImpl(Workspace *parent);

    ~WorkspaceImpl() = default;

    // Use this specific changelist number instead of the default changelist
    // for calls to p4 reconcile.
    void
    setChangelist(const std::string & c);

    // Use this specific changelist number instead of the default changelist
    // for calls to p4 reconcile.
    const std::string &
    changelist() const;

    // The time we wait from the last file operation to trigger p4 reconcile.
    int
    pauseAmount() const;

    // The time we wait from the last file operation to trigger p4 reconcile.
    void setPauseAmount(int t);

    // If true, the workspace is churning
    bool
    processing() const;

    // Debugging status for the last operations
    std::string
    status();

    void
    setStatus(const std::string & s);

    void
    enqueue(const Notification & n);

    template<typename Lambda>
    int execute(Lambda lambda);

private: // Methods that should never ever see the light of day

    void startCheckTimer();

    void check();

    void startProcessTimer();

    void process();

private:
    Workspace *_parent;
    string _changelist;
    int _pauseAmount;
    Timer _timer;

    mutex _statusMutex;
    string _status;

    Timer::timer_id _checkTimer;

    Timer::timer_id _processTimer;

    // We maintain a simple vector of Notification instances, and lock access
    // when we attempt to add or clear the vector.
    mutex _notificationMutex;
    vector<Notification> _notifications;
};


WorkspaceImpl::WorkspaceImpl(Workspace *parent)
: _parent(parent), _changelist(), _pauseAmount(500), _timer(),
  _statusMutex(), _status("ok"), _checkTimer(-1), _processTimer(-1),
  _notificationMutex(), _notifications()
{
}


void
WorkspaceImpl::setChangelist(const std::string & c)
{
    _changelist = c;
}

const std::string &
WorkspaceImpl::changelist() const
{
    return _changelist;
}

int
WorkspaceImpl::pauseAmount() const
{
    return _pauseAmount;
}

void
WorkspaceImpl::setPauseAmount(int t)
{
    _pauseAmount = t;
}

bool
WorkspaceImpl::processing() const
{
    return _checkTimer != -1 && _processTimer != -1;
}

std::string
WorkspaceImpl::status()
{
    _statusMutex.lock();
    string s = _status;
    _statusMutex.unlock();

    return s;
}

void
WorkspaceImpl::setStatus(const std::string & s)
{
    _statusMutex.lock();
    _status = s;
    _statusMutex.unlock();
}

void
WorkspaceImpl::enqueue(const Notification & n)
{
    _notificationMutex.lock();
    _notifications.push_back(n);
    _notificationMutex.unlock();

    if (!processing())
        startCheckTimer();
}

void
WorkspaceImpl::startCheckTimer()
{
    _checkTimer = _timer.create(pauseAmount(), 0, [this]() { this->check(); });
}

void
WorkspaceImpl::check()
{
    _checkTimer = -1;

    _notificationMutex.lock();
    if ((_notifications.back().time + milliseconds(_pauseAmount)) < system_clock::now())
    {
        _notifications.clear();
        startProcessTimer();
    }
    else
    {
        // Come back later
        startCheckTimer();
    }
    _notificationMutex.unlock();
}

void
WorkspaceImpl::startProcessTimer()
{
    _processTimer = _timer.create(0, 0, [this]() { this->process(); });
}

void
WorkspaceImpl::process()
{
    execute([this](ClientApiHandle & handle) {
        CommandResult result = handle.run("reconcile");
        if (result.hasError())
        {
            stringstream ss;
            for (auto message : result.messages())
            {
                ss << message.text << endl;
            }
            setStatus(ss.str());
        }
        else
        {
            setStatus("ok");
        }
        return 0;
    });

    _processTimer = -1;
}

// Basically handles connecting and disconnecting the ClientApiHandle, which
// creates our initial approach to connection management.
//
// The Lambda should be a function that returns an int and takes the
// ClientApiHandle as a reference.
template<typename Lambda>
int
WorkspaceImpl::execute(Lambda lambda)
{
    shared_ptr<ClientApiHandle> apiHandle(
    ClientApiHandle::create(_parent->workspace()));

    apiHandle->api().SetProtocol("tag", "");

    Error e;

    apiHandle->api().Init(&e);

    if (e.Test())
    {
        StrBuf strBuf;
        e.Fmt(&strBuf);
        cerr << strBuf.Text() << endl;
        return Workspace::CONNECTION_FAILURE;
    }

    apiHandle->setConnected(true);

    return lambda(*apiHandle);
}

//----------------------------------------------------------------------------
// Workspace
//----------------------------------------------------------------------------

Workspace::~Workspace()
{
    delete _impl;
}

Workspace::Workspace()
: _workspace(), _impl(new WorkspaceImpl(this))
{
}

Workspace::Workspace(const std::string & workspace)
: _workspace(workspace), _impl(new WorkspaceImpl(this))
{
}

std::string
Workspace::path(const std::string & p) const
{
    // This actually works. FUSE just sends in paths that always start with "/".
    return _workspace + p;
}


void
Workspace::setChangelist(const std::string & c)
{
    _impl->setChangelist(c);
}

const std::string &
Workspace::changelist() const
{
    return _impl->changelist();
}

int
Workspace::pauseAmount() const
{
    return _impl->pauseAmount();
}

void
Workspace::setPauseAmount(int t)
{
    _impl->setPauseAmount(t);
}

bool
Workspace::processing() const
{
    return _impl->processing();
}

std::string
Workspace::status()
{
    return _impl->status();
}

void
Workspace::enqueue(const Notification & n)
{
    _impl->enqueue(n);
}