// -*- Mode: C++; -*-
//                              File      : FilterAdmin_i.cc
//                              Package   : omniNotify-Library
//                              Created on: 1-Jan-1998
//                              Authors   : gruber&panagos
//
//    Copyright (C) 1998-2001 AT&T Laboratories -- Research
//
//    This file is part of the omniNotify library
//    and is distributed with the omniNotify release.
//
//    The omniNotify library is free software; you can redistribute it and/or
//    modify it under the terms of the GNU Library General Public
//    License as published by the Free Software Foundation; either
//    version 2 of the License, or (at your option) any later version.
//
//    This library 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
//    Library General Public License for more details.
//
//    You should have received a copy of the GNU Library General Public
//    License along with this library; if not, write to the Free
//    Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA
//    02111-1307, USA
//
//
// Description:
//    Implementation of FilterFactory_i and FAdminHelper
//
 
/*
  $Log: FilterAdmin_i.cc,v $
  Revision 1.42  2001/10/02 14:07:52  alcfp
  improved error messages for command line errors

  Revision 1.41  2001/09/28 02:10:49  alcfp
  FilterFactory_i uses oplockptr

  Revision 1.40  2001/09/19 21:10:04  alcfp
  Added cleanup support to interactive api

  Revision 1.39  2001/09/07 00:56:57  alcfp
  Filter_i uses filter id to track instances.  Added cleanup command to FilterFactory

  Revision 1.38  2001/09/05 18:02:44  alcfp
  more interactive commands working

  Revision 1.37  2001/08/26 16:09:32  alcfp
  more interactive stuff working

  Revision 1.36  2001/08/03 17:54:17  alcfp
  added support for AttNotification

  Revision 1.35  2001/06/26 20:01:16  alcfp
  updated copyright notices, added support for omniORB4, switched default to POA

  Revision 1.34  2001/06/22 07:00:33  alcfp
  moved to new logging scheme

  Revision 1.33  2000/11/15 21:17:29  alcfp
  large number of changes to switch to use of RDIOplocks for safe object disposal support.  also reduced code duplication a little, and tried hard to make all the proxy code consistent

  Revision 1.32  2000/08/22 18:23:53  alcfp
  added description to each file

  Revision 1.31  2000/08/16 20:19:29  alcfp
  Added licensing notice to each .h and .cc file where library files get GLPL notice and daemon file gets GPL notice -- examples do not claim any license but point out that the library and daemon code does have a license notice

*/
 
#include "RDI.h"
#include "RDINotifServer.h"
#include "RDIStringDefs.h"
#include "RDIList.h"
#include "RDICatchMacros.h"
#include "RDIOplocksMacros.h"
#include "CosNfyUtils.h"
#include "CosNotifyFilter_i.h"
#include "CosNotifyChannelAdmin_i.h"

////////////////////////////////////////////////////////////////////

FilterFactory_i::FilterFactory_i(const char* grammar) :
  _oplockptr(0), _disposed(0), _nlangs(0)
{
  RDI_OPLOCK_INIT;
  _my_name.length(2);
  _my_name[0] = (const char*)"server";
  _my_name[1] = (const char*)"filtfact";
  for (unsigned int i=0; i < MAXGR; i++) {
    _clangs[i] = 0; 
  }
  if ( ! (_clangs[0] = CORBA_STRING_DUP(grammar)) ) {
    throw CORBA::NO_MEMORY(0, CORBA::COMPLETED_NO);
  }
  _nlangs += 1;
  WRAPPED_REGISTER_IMPL2(this, &_my_name);
}

FilterFactory_i::~FilterFactory_i()
{
  RDI_OPLOCKS_DESTROY_CHECK("FilterFactory_i");
}

void
FilterFactory_i::cleanup_and_dispose()
{
  RDI_OPLOCK_ACQUIRE(return);
  if (_disposed) {
    RDIDbgFAdminLog("FilterFactory_i::cleanup_and_dispose() called more than once\n");
    RDI_OPLOCK_RELEASE;
    return;
  }
  _disposed = 1; // acts as guard: the following is executed by only one thread 
  for (unsigned int i=0; i < MAXGR; i++) {
    CORBA_STRING_FREE(_clangs[i]);
    _clangs[i] = 0;
  }
  _nlangs  = 0;
  RDI_OPLOCK_DISPOSE_IMPL_AND_RELEASE;
}

CosNF::Filter_ptr
FilterFactory_i::create_filter(const char* grammar  WRAPPED_IMPLARG )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  Filter_i* fltr = 0;
  RDIDbgFAdminLog("Create a new filter using: " << grammar << '\n');
  if ( ! _is_supported(grammar) ) {
    RDI_OPLOCK_RELEASE;
    throw CosNF::InvalidGrammar();
  }
  if ( ! (fltr = new Filter_i(grammar, this)) ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::NO_MEMORY(0, CORBA::COMPLETED_NO);
  }
  RDI_AssertAllocThrowNo(fltr, "Memory allocation failure -- Filter_i\n");
  RDI_OPLOCK_RELEASE;
  return WRAPPED_IMPL2OREF(CosNF::Filter, fltr);
}

CosNF::MappingFilter_ptr
FilterFactory_i::create_mapping_filter(const char* grammar, const CORBA::Any& value   WRAPPED_IMPLARG )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  MappingFilter_i* fltr = 0;
  if ( ! _is_supported(grammar) ) {
    RDI_OPLOCK_RELEASE;
    throw CosNF::InvalidGrammar();
  }
  if ( ! (fltr = new MappingFilter_i(grammar, value, this)) ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::NO_MEMORY(0, CORBA::COMPLETED_NO);
  }
  RDIDbgFAdminLog("Created new MappingFilter using: " << grammar << '\n');
  RDI_OPLOCK_RELEASE;
  return WRAPPED_IMPL2OREF(CosNF::MappingFilter, fltr); 
}

int
FilterFactory_i::add_grammar(const char* grammar)
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if ( _is_supported(grammar) ) {
    RDI_OPLOCK_RELEASE;
    return 0;
  }
  if ( _nlangs == MAXGR ) {
    RDI_OPLOCK_RELEASE;
    RDIDbgFAdminLog("reached max number of supported grammars\n");
    return -1;
  }
  for (unsigned int i=0; i < MAXGR; i++) {
    if ( _clangs[i] == (char *) 0 ) {
      if ( ! (_clangs[i] = CORBA_STRING_DUP(grammar)) ) {
	RDI_OPLOCK_RELEASE;
	RDIDbgFAdminLog("failed to allocate string for " << grammar << '\n');
	return -1;
      } else {
	RDIDbgFAdminLog("Adding support for: " << grammar << '\n');
	_nlangs += 1;
	RDI_OPLOCK_RELEASE;
	return 0;
      }
    }
  }
  RDI_OPLOCK_RELEASE;
  RDIDbgForceLog("Internal error -- inconsistent data structures.....\n");
  return -1;
}

void
FilterFactory_i::del_grammar(const char* grammar)
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  for (unsigned int i=0; i < MAXGR; i++) {
    if ( _clangs[i] && RDI_STR_EQ(_clangs[i], grammar) ) {
      RDIDbgFAdminLog("Deleting support for: " << grammar << '\n');
      CORBA_STRING_FREE(_clangs[i]);
      _nlangs -= 1;
      break;
    }
  }
  RDI_OPLOCK_RELEASE;
}

CORBA::Boolean
FilterFactory_i::is_supported(const char* grammar)
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  CORBA::Boolean res = _is_supported(grammar);
  RDI_OPLOCK_RELEASE;
  return res;
}

// oplock already acquired; does the real work
CORBA::Boolean
FilterFactory_i::_is_supported(const char* grammar)
{
  for (unsigned int i=0; i < MAXGR; i++) {
    if ( _clangs[i] && RDI_STR_EQ(_clangs[i], grammar) ) {
      return 1;
    }
  }
  return 0;
}

////////////////////////////////////////
// Interactive

AttN::NameSeq*
FilterFactory_i::my_name( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  AttN::NameSeq* res = new AttN::NameSeq(_my_name);
  if ( res == (AttN::NameSeq*) 0 ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::NO_MEMORY(0, CORBA::COMPLETED_NO);
  }
  RDI_OPLOCK_RELEASE;
  return res;
}

AttN::NameSeq*
FilterFactory_i::child_names( WRAPPED_IMPLARG_VOID )
{
  return Filter_i::all_filter_names();
}

AttN::IactSeq*
FilterFactory_i::children(CORBA::Boolean only_cleanup_candidates WRAPPED_IMPLARG )
{
  return Filter_i::all_children(only_cleanup_candidates);
}

CORBA::Boolean
FilterFactory_i::safe_cleanup( WRAPPED_IMPLARG_VOID )
{
  return 0; // not destroyed
}

void
FilterFactory_i::cleanup_all(RDIstrstream& str)
{
  str << "\nDestroying all filters with zero callbacks\n";
  unsigned int num_destroyed = 0;
  AttN::IactSeq* cleanups = Filter_i::all_children(1);
  if (cleanups) {
    for (unsigned int i = 0; i < cleanups->length(); i++) {
      AttN::NameSeq* nm = 0;
      CORBA::Boolean worked = 0, prob = 0;
      try {
	nm = (*cleanups)[i]->my_name();
	worked = (*cleanups)[i]->safe_cleanup();
      } CATCH_INVOKE_PROBLEM(prob);
      if (worked) {
	num_destroyed++;
	str << "Destroyed filter " << *nm << '\n';
      }
      if (nm) delete nm;
    }
    delete cleanups;
  }
  str << "Total filters destroyed: " << num_destroyed <<  '\n';
}

void
FilterFactory_i::out_commands(RDIstrstream& str)
{
  str << "omniNotify FilterFactory commands:\n"
      << "  cleanup           : destroy all filters that have zero callbacks\n"
      << "                        (normally means a client forgot to destroy a filter;\n"
      << "                         sometimes filter not yet added to a proxy or admin)\n"
      << "  info filters      : show brief description of each non-destroyed filter\n"
      << "  info <filt_name>  : show description of a specific filter\n"
      << "  up                : change target to server\n"
      << "  go <filt_name>    : change target to a specific filter\n"
      << "                        ('children' lists filter names)\n";
}

char*
FilterFactory_i::do_command(const char* cmnd, CORBA::Boolean& success,
			    CORBA::Boolean& target_changed,
			    AttN_Interactive_outarg next_target  WRAPPED_IMPLARG )
{
  RDIParseCmd p(cmnd);
  success = 1;
  target_changed = 0;
  if (p.argc == 0) { return CORBA_STRING_DUP("");  }

  RDINotifServer* server = RDI::get_server_i();
  RDIstrstream str;
  if ((p.argc == 1) && RDI_STR_EQ_I(p.argv[0], "help")) {
    out_commands(str);
  } else if ((p.argc == 1) && RDI_STR_EQ_I(p.argv[0], "up")) {
    target_changed = 1;
    next_target = RDI::get_server();
    str << "\nomniNotify: new target ==> server\n";
  } else if ((p.argc == 1) && RDI_STR_EQ_I(p.argv[0], "cleanup")) {
    cleanup_all(str);
  } else if ((p.argc == 2) &&
	     RDI_STR_EQ_I(p.argv[0], "info") && RDI_STR_EQ_I(p.argv[1], "filters")) {
    Filter_i::out_info_all_filters(str);
  } else if ((p.argc == 2) &&
	     RDI_STR_EQ_I(p.argv[0], "info")) {
    Filter_i::out_info_filter(str, p.argv[1]);
  } else if ((p.argc == 2) &&
	     RDI_STR_EQ_I(p.argv[0], "go")) {
    Filter_i* f = Filter_i::find_filter(p.argv[1]);
    if (f) {
      target_changed = 1;
      next_target = WRAPPED_IMPL2OREF(AttN::Filter, f);
      str << "\nomniNotify: new target ==> " << p.argv[1] << '\n';
    } else {
      str << "Invalid target: " << p.argv[1] << " is not a filter name\n";
      str << "  (Use 'children' for list of valid filter names)\n";
      success = 0;
    }
  } else {
    str << "Invalid command: " << cmnd << "\n";
    success = 0;
  }
  RDIRptForceLog(_my_name << " received command: " << cmnd << "\nResult:\n" << str.buf());
  // this is the only safe way to return a string?
  return CORBA_STRING_DUP(str.buf());
}

////////////////////////////////////////////////////////////////////

FAdminHelper::FAdminHelper() : _oplock(), _serial(1),
			       _rdcbids(RDI_ULongHash, RDI_ULongRank),
			       _filters(RDI_ULongHash, RDI_ULongRank) {;}

FAdminHelper::~FAdminHelper()
{ remove_all_filters(); }

CosNF::FilterID FAdminHelper::add_filter_i(CosNF::Filter_ptr new_filter,
					   RDINotifySubscribe_ptr filter_holder)
{
  CosNF::FilterID   fltrID;
  CosNF::CallbackID rdcbID;
  Filter_i* fltr = Filter_i::Filter2Filter_i(new_filter);
  RDI_Assert(fltr, "Filter was not created by READY\n"); 
  omni_mutex_lock lock(_oplock);
  fltrID = _serial++;
  if ( _filters.insert(fltrID, fltr) ) {
    RDIDbgFAdminLog("Failed to register new filter in hash table\n");
    return 0;
  }
  CosNF::Filter::_duplicate(new_filter);
  rdcbID = fltr->attach_callback_i(filter_holder);
  if ( _rdcbids.insert(fltrID, rdcbID) ) {
    RDIDbgFAdminLog("Failed to register new callback ID in hash table\n");
    _filters.remove(fltrID);
    return 0;
  }
  RDIDbgFAdminLog("\tFilter " << new_filter << " [" << fltrID << "] for " << filter_holder << '\n');
  return fltrID;
}

// ******************************************************************** //
// The following is invoked by SupplierAdmin and ProxyConsumer objects. //
// Filters attached to such objects are treated differently than normal //
// consumer filters.  In particular,  changes in event types referenced //
// in these filters are NOT propagated to suppliers.                    //
// ******************************************************************** //

CosNF::FilterID FAdminHelper::add_filter_i(CosNF::Filter_ptr new_filter)
{
  CosNF::FilterID fltrID;
  Filter_i* fltr = Filter_i::Filter2Filter_i(new_filter);
  RDI_Assert(fltr, "Filter was not created by READY\n");
  omni_mutex_lock lock(_oplock);
  fltrID = _serial++;
  if ( _filters.insert(fltrID, fltr) ) {
    RDIDbgFAdminLog("Failed to register new filter in hash table\n");
    return 0;
  }
  CosNF::Filter::_duplicate(new_filter);
  return fltrID;
}

// ---------------------------------------------------------------- //
// A filter is being destroyed and it is registered with the object //
// that uses this particular FAdminHelper object. To avoid run-time //
// errors, we should remove the filter from the internal hash table //
// ---------------------------------------------------------------- //

void FAdminHelper::rem_filter_i(Filter_i* fltr)
{
  RDI_HashCursor<CosNF::FilterID, Filter_i*> curs;
  omni_mutex_lock lock(_oplock);
  for (curs = _filters.cursor(); curs.is_valid(); ++curs) {
    Filter_i*& fval = curs.val();
    if ( fval->getID() == fltr->getID() ) {
      RDIDbgFAdminLog("removing filter [" << curs.key() << "] from hash table\n");
      _filters.remove( curs.key() );
      _rdcbids.remove( curs.key() );
      break;
    }
  }
}

CosNF::Filter_ptr FAdminHelper::get_filter(CosNF::FilterID fltrID 
					   WRAPPED_IMPLARG )
{
  Filter_i*       fltr=0;
  omni_mutex_lock lock(_oplock);
  if ( _filters.lookup(fltrID, fltr) ) {
    RDIDbgFAdminLog("Get Filter " << fltr << " [" << fltrID << "]\n");
    return WRAPPED_IMPL2OREF(CosNF::Filter, fltr);
  }
  throw CosNF::FilterNotFound();
}

CosNF::FilterIDSeq* FAdminHelper::get_all_filters( WRAPPED_IMPLARG_VOID )
{
  RDI_HashCursor<CosNF::FilterID, Filter_i*> fcur;
  CORBA::ULong  indx=0;
  CosNF::FilterIDSeq* fseq=new CosNF::FilterIDSeq();
  RDI_AssertAllocThrowNo(fseq, "Memory allocation failure -- FilterIDSeq\n");
  omni_mutex_lock lock(_oplock);
  fseq->length(_filters.length());
  for ( fcur = _filters.cursor(); fcur.is_valid(); fcur++, indx++ ) {
    (*fseq)[indx] = fcur.key();
  }
  return fseq;
}

void FAdminHelper::remove_filter(CosNF::FilterID fltrID
				 WRAPPED_IMPLARG )
{
  Filter_i* fltr=0;
  omni_mutex_lock lock(_oplock);
  if ( ! _filters.lookup(fltrID, fltr) ) {
    throw CosNF::FilterNotFound();
  }
  // cleanup_filter(fltrID);
  _filters.remove(fltrID);
  _rdcbids.remove(fltrID);
  WRAPPED_RELEASE_IMPL(fltr);
}

void FAdminHelper::remove_all_filters( WRAPPED_IMPLARG_VOID )
{
  RDI_HashCursor<CosNF::FilterID, Filter_i*> fcur;
  omni_mutex_lock lock(_oplock);
  for ( fcur = _filters.cursor(); fcur.is_valid(); fcur++ ) {
    Filter_i*& fltr = fcur.val();
    cleanup_filter(fcur.key());
  }
  _rdcbids.clear();
  _filters.clear();
}

void FAdminHelper::cleanup_filter(CosNF::FilterID fltrID)
{
  CosN::EventTypeSeq addseq;
  CosN::EventTypeSeq delseq;
  CosNF::CallbackID  rdcbID;
  Filter_i* fltr=0;

  _filters.lookup(fltrID, fltr);
  _rdcbids.lookup(fltrID, rdcbID);
  addseq.length(0);
  delseq.length(0);

  // We have to notify all interested parties about the deletion of
  // this filter. In particular, we need to collect all event types
  // referenced in the constraints of the filter and propagate them
  // to the subscribers of 'subscription_change'

  fltr->create_ev_types_from_dom_list(delseq);
  fltr->notify_subscribers_i(addseq, delseq);
  fltr->detach_callback_i(rdcbID);

  WRAPPED_RELEASE_IMPL(fltr);
}

void FAdminHelper::out_info_filters(RDIstrstream& str) {
  RDI_HashCursor<CosNF::FilterID, Filter_i*> fcur;
  omni_mutex_lock lock(_oplock);
  if (_filters.length() == 0) {
    str << "  (no attached filters)\n";
  } else {
    for ( fcur = _filters.cursor(); fcur.is_valid(); fcur++ ) {
      fcur.val()->out_info_descr(str);
    }
  }
}

