text: /*
     Copyright (C) 2003, 2005, 2006 MySQL AB
      All rights reserved. Use is subject to license terms.
  
     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; version 2 of the License.
  
     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, write to the Free Software
     Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
  */
  
  #include "MemoryChannel.hpp"
  #include "NdbThread.h"
  #include "NdbSleep.h"
  #include "NdbOut.hpp"
  #include "NdbMain.h"
  
  
  
  MemoryChannel<int>* theMemoryChannel;
  
  
  extern "C" void* runProducer(void*arg)
  {
    // The producer will items into the MemoryChannel
    int count = *(int*)arg;
    int* p;
    int i = 0;
    while (i <= count)
      {
        p = new int(i);
        ndbout << "P: " << *p << endl;
        theMemoryChannel->writeChannel(p);
        if (i%5==0)
          NdbSleep_MilliSleep(i);
        i++;
      }
    return NULL;
  }
  
  extern "C" void* runConsumer(void* arg)
  {
    // The producer will read items from MemoryChannel and print on screen
    int count = *(int*)arg;
    int* p;
    int i = 0;
    while (i < count)
      {
        p = theMemoryChannel->readChannel();
        ndbout << "C: " << *p << endl;
        i = *p;
        delete p;
        
      }
    return NULL;
  }
  
  
  
  class ArgStruct 
  {
  public:
    ArgStruct(int _items, int _no){
      items=_items; 
      no=_no;
    };
    int items;
    int no;
  };
  
  MemoryChannelMultipleWriter<ArgStruct>* theMemoryChannel2;
  
  extern "C" void* runProducer2(void*arg)
  {
    // The producer will items into the MemoryChannel
    ArgStruct* pArg = (ArgStruct*)arg;
    int count = pArg->items;
    ArgStruct* p;
    int i = 0;
    while (i < count)
      {
        p = new ArgStruct(i, pArg->no);
        ndbout << "P"<<pArg->no<<": " << i << endl;
        theMemoryChannel2->writeChannel(p);
        NdbSleep_MilliSleep(i);
        i++;
      }
    return NULL;
  }
  
  extern "C" void* runConsumer2(void* arg)
  {
    // The producer will read items from MemoryChannel and print on screen
    ArgStruct* pArg = (ArgStruct*)arg;
    int count =  pArg->items * pArg->no;
    ArgStruct* p;
    int i = 0;
    while (i < count)
      {
        p = theMemoryChannel2->readChannel();
        ndbout << "C: "<< p->no << ", " << p->items << endl;
        i++;
        delete p;
      }
    ndbout << "Consumer2: " << count << " received" << endl;
    return NULL;
  }
  
  
  
  
  //#if defined MEMORYCHANNELTEST
  
  //int main(int argc, char **argv)
  NDB_COMMAND(mctest, "mctest", "mctest", "Test the memory channel used in Ndb", 32768)
  {
  
    ndbout << "==== testing MemoryChannel ====" << endl;
  
    theMemoryChannel = new MemoryChannel<int>;
    theMemoryChannel2 = new MemoryChannelMultipleWriter<ArgStruct>;
  
    NdbThread* consumerThread;
    NdbThread* producerThread;
  
    NdbThread_SetConcurrencyLevel(2);
  
    int numItems = 100;
    producerThread = NdbThread_Create(runProducer, 
  				    (void**)&numItems,
  				    4096,
  				    (char*)"producer");
  
    consumerThread = NdbThread_Create(runConsumer, 
  				    (void**)&numItems,
  				    4096,
  				    (char*)"consumer");
  
  
    void *status;
    NdbThread_WaitFor(consumerThread, &status);
    NdbThread_WaitFor(producerThread, &status);
  
    ndbout << "==== testing MemoryChannelMultipleWriter ====" << endl;
  #define NUM_THREADS2 5
    NdbThread_SetConcurrencyLevel(NUM_THREADS2+2);
    NdbThread* producerThreads[NUM_THREADS2];
  
    ArgStruct *pArg;
    for (int j = 0; j < NUM_THREADS2; j++)
      {
        char buf[25];
        sprintf((char*)&buf, "producer%d", j);
        pArg = new ArgStruct(numItems, j);
        producerThreads[j] = NdbThread_Create(runProducer2, 
  				    (void**)pArg,
  				    4096,
  				    (char*)&buf);
      }
  
    pArg = new ArgStruct(numItems, NUM_THREADS2);
    consumerThread = NdbThread_Create(runConsumer2, 
  				    (void**)pArg,
  				    4096,
  				    (char*)"consumer");
  
  
    NdbThread_WaitFor(consumerThread, &status);
    for (int j = 0; j < NUM_THREADS2; j++)
    {
      NdbThread_WaitFor(producerThreads[j], &status);
    }
  
  				    
    return 0;
  
  }
  
  void ErrorReporter::handleError(ErrorCategory type, int messageID,
                                  const char* problemData, const char* objRef,
  				NdbShutdownType nst)
  {
  
    ndbout << "ErrorReporter::handleError activated"  << endl;
    exit(1);
  }
  
  //#endif
  