/* "mani" -- Multi-way manifold for ppipc streams */

/********************************************************************
 * Copyright 1990 by Peter Goodeve                                  *
 * This program may be distributed and used in any way whatsoever,  *
 * provided that this copyright notice remains intact, and that     *
 * no charge, aside from a small fee to cover duplication and       *
 * distribution costs, is made for this.                            *
 ********************************************************************/
/* 90:11:18 */

#include <exec/types.h>
#include <proto/exec.h>
#include <libraries/dos.h>
#include <proto/dos.h>
#include <string.h>
#include <stdio.h>

#include "IPC.h"
#include "IPC_proto.h"

#define MAX_ITEMS 4


/* Message IDs: (only one needed!) */


#define QUIT  MAKE_ID('Q','U','I','T')


/************************************************************/

#ifdef BACKGND

long _stack = 4000;
char * _procname = "mani";
long _priority = 0;    /* making this higher fouls up gadget highlighting! */
long _BackGroundIO = 1;
extern BPTR _Backstdout;

#endif

struct Library * IPCBase = NULL; /* Base pointer for the IPC Library */


struct IPCPort * source, * dest[20];
struct MsgPort * rport;

ULONG s_sig, r_sig;

struct IPCMessage * msgs[20];
int dcount;

int msg_limit, item_limit;

int outstanding;    /* number of messages sent but not yet returned */
int autoquit = TRUE;    /* by default quit when all clients do */


/***************************************************************/

void DisplayLine(char * str); /* to avoid puts() (for backgrounding) */

void DisposeIPCMsg(struct IPCMessage * msg)
{
    if (msg->ipc_Msg.mn_ReplyPort) ReplyIPCMessage(msg);
    else DeleteIPCMsg(msg); /* any data better be inline or already gone! */
}

/***************************************************************/

int valid_msg(struct IPCMessage * msg)
{
    int i;
    struct IPCItem *item;
    if ((i = msg->ipc_ItemCount) > item_limit) return FALSE;
    if (msg->ipc_Msg.mn_Length > msg_limit) return FALSE;
    for (item = msg->ipc_Items; i; item++, i--)
        if (item->ii_Flags & IPC_TRANSFER) {
            item->ii_Flags |= IPC_REJECT;
            msg->ipc_Flags |= IPC_REJECT | IPC_CHECKITEM;
            return FALSE;
        }
    return TRUE;
}

void dupl_msg(struct IPCMessage * msg)
{
    int i, icount;
    struct IPCMessage ** mp = msgs;
    struct IPCPort ** dp = dest;
    for (i=dcount; i--;  mp++, dp++ ) {
        struct IPCItem *sitem = msg->ipc_Items,
                       *ditem = (*mp)->ipc_Items;
        for (icount = msg->ipc_ItemCount;  icount;
              sitem++, ditem++, icount--) *ditem = *sitem;
        (*mp)->ipc_ItemCount = msg->ipc_ItemCount;
        (*mp)->ipc_Flags = msg->ipc_Flags;
        (*mp)->ipc_Id = msg->ipc_Id;
        if (PutIPCMsg(*dp, *mp)) outstanding++;
    }
}

void handle_replies()
{
    do {
        while (GetMsg(rport)) outstanding--; /* just keep track of replies */
        if (outstanding <= 0) break; /*BETTER never be negative, actually! */
        Wait(r_sig); /* could use WaitPort instead... */
    } while (TRUE);
    return; /* when eveything is back */
}

void source_loop()
{
    struct IPCMessage *smsg;
    int running = TRUE;
    do {
        while (smsg = GetIPCMessage(source)) {
            if (valid_msg(smsg)) {
                dupl_msg(smsg);
                handle_replies();
            }
            if (smsg->ipc_Id == QUIT) {
                        /* after we've told everyone else to */
                running = FALSE;
            }
            DisposeIPCMsg(smsg);
        }
        if (running) Wait(s_sig); /** canNOT use WaitPort here, because of: */
        if (autoquit && CheckIPCPort(source, IPP_NOTIFY) == 1) break;
    } while (running);
}


/************************************************************/

void Cleanup() {
    if (source) {
        struct IPCMessage * msg;
        ShutIPCPort(source);
        while (msg = GetIPCMessage(source)) {
            msg->ipc_Flags |= IPC_REJECT;
            DisposeIPCMsg(msg);
        }
        LeaveIPCPort(source);
    }
    while (dcount--) {
        DeleteIPCMsg(msgs[dcount]);
        DropIPCPort(dest[dcount]);
    }
    if (rport) LeaveIPCPort(rport);
    if (IPCBase) CloseLibrary(IPCBase);
#ifdef BACKGND
        if (_Backstdout)
            Close(_Backstdout);
        _Backstdout = NULL;
#endif
}


/************************************************************/

void main(int argc, char ** argv)
{
    if (argc < 3) {
        DisplayLine("...must have an input and at least one output");
        Cleanup();
        exit(22);
    }
    if (argc > 22) {
        DisplayLine("...unable to handle more than 20 outputs");
        Cleanup();
        exit(22);
    }

    IPCBase = OpenLibrary("ppipc.library",0);
    if (!IPCBase) {
        DisplayLine(
        "Couldn't find ppipc.library!\nHave you installed it in LIBS: ?");
        Cleanup();
        exit(20);
    }

    /* create a port for replies to distributed messages: */
    rport = ServeIPCPort(NULL);    /* actually a std MsgPort, but
                                      this is convenient */
    if (!rport) {
        DisplayLine("Insufficient memory");
        Cleanup();
        exit(22);
    }
    r_sig = SigBitIPCPort(rport);


    /* Become a Server for the specified source port name: */
    if (*argv[1] == '!') { /* same convention as IP: device */
        autoquit = FALSE;
        argv[1]++;  /* move past flag char */
    }
    source = ServeIPCPort(argv[1]);
    if (!source) {
        DisplayLine("Couldn't link to source");
        Cleanup();
        exit(22);
    }
    s_sig = SigBitIPCPort(source);

#ifdef BACKGND
        /* now we can drop completely into the background */
        if (_Backstdout)
            Close(_Backstdout);
        _Backstdout = NULL;
#endif
    /* Link to all the other argument-line ports as a Client */
    /* and create a single message for each (to be used repeatedly): */
    argc -= 2; /* ignore commandname & source */
    argv += 2;
    for ( dcount = 0; argc; argc--, argv++, dcount++) {
        dest[dcount] = GetIPCPort(*argv);
        msgs[dcount] = CreateIPCMsg(MAX_ITEMS, 0, rport);
                                           /* fixed limits for now */
    }
    msg_limit = (msgs[0])->ipc_Msg.mn_Length; /* record for validity check */
    item_limit = MAX_ITEMS;

    source_loop(); /* go and loop until source is closed or QUIT received */

    Cleanup();
}

/************************************************************/

void DisplayLine(char * str) /* to avoid puts() */
{
#ifdef BACKGND
    if (_Backstdout)
         Write(_Backstdout, str, strlen(str));
         Write(_Backstdout, "\n", 1);
#else
    Write(Output(), str, strlen(str));
    Write(Output(), "\n", 1);
#endif
}