Hi all,

I am having a little problem with my relay buffer which uses flow design patterns. The problem is that when the consumer should output the value of the Element, it is printing out "The item is: null". My code is a little bit long, but hopefully you can bare with me, and help me find out why it comes out as null:

// Buffer.java

public interface Buffer extends Put, Take
{

            public void put ( Element item ) ;

            public Element take ( ) ;

   

}
// DummyBuffer.java

class DummyBuffer implements Buffer
{
    private Element contents ;
    private boolean available = false;

    public synchronized Element take( )
    {
	while ( !available )
	{
	    try {
		   wait();
		}
	    catch ( InterruptedException e ) { }
	}

	available = false ;

	notifyAll( ) ;

	return contents ;
    }


    public synchronized void put( Element element )
    {
	while ( available )
	{
	    try {
		    wait( ) ;
	    }
	    catch ( InterruptedException e ) { }
	}

	contents = element ;

	available = true ;

	notifyAll() ;

    }
}
// Element.java

class Element extends Object
{

    private Object element ;

    public Element( )
    {

	super() ;

	element = new String( "This is an Element" ) ;

    }

    public Element( Object e )
    {

	super() ;

	element = e ;

    }


    public String toString()
    {

	return element.toString() ;

    }

}
// EndOfData.java

class EndOfData extends Element
{

    private final String DEFAULT_MESSAGE = "TERMINATE" ;

    private final String message ;

    //  default message version
    EndOfData ( )
    {
	message = DEFAULT_MESSAGE ;
    }

    // message version
    EndOfData( String message )
    {
	this.message = message ;
    }

    public String toString ()
    {
	return "EndOfData[" + message + "]" ;
    }

}
// LinearStage.java

abstract class LinearStage extends Stage
{

    // 1 "take" input source
    private final TakeStage in  ;

    // 1 "put" output source
    private final PutStage  out ;


    public LinearStage( TakeStage in, PutStage out )
    {
	// Stage constructor
	super() ;

	this.in  = in  ;
	this.out = out ;
    }


    public void run ( )
    {

	Element element = in.take() ;

	// while not end of data retrieve more items

	while ( notEndOfData( element ) )
	    {
		Element processedElement = processData( element ) ;

		out.put( processedElement ) ;

		element = in.take() ;

	    }


	// propagate the "EndOfData"

	out.put( new EndOfData() ) ;

    }


    // Abstract methods

    // used to indicate when the stage should terminate

    abstract protected boolean notEndOfData( Element element ) ;


    // used to ``process'' the data

    abstract protected Element processData( Element element ) ;


}
// Put.java

public interface Put
{
    abstract public void put( Element item ) ;

}
// PutStage.java

abstract public class PutStage extends    Stage
		               implements Put
{


    public PutStage( )
    {
	// Stage constructor
	super() ;

    }
}
// RB_Consumer.java

class RB_Consumer extends PutStage
{

    private Element item ;


    public RB_Consumer( )
    {
	// PutStage's constructor
	super() ;

    }

    public synchronized void put( Element item )
    {

	this.item = item ;

    }


    public void run()
    {

	// use `item'
	System.out.println( "The item: " + item ) ;

    }

}
// RB_Producer.java

class RB_Producer extends Stage
{

    private Put nextStage ;

    public RB_Producer(Put nextStage )
    {

	super() ;

	this.nextStage = nextStage ;

    }



    public void run( )
    {

	Element item = new Element( "An Element" ) ;
   
	nextStage.put( item ) ;

    }

}
// RelayBuffer.java

public class RelayBuffer implements Runnable,
				    Put
{

 
    protected Put consumer ;

 
    protected Buffer buffer = new DummyBuffer() ;

    protected Thread pusher ;


    public RelayBuffer( PutStage consumer )
    {

	this.consumer = consumer ;

	// create & start the ``pusher'' thread

	pusher = new Thread(this) ;
	pusher.start() ;

    }




    public synchronized void put( Element e )
    {
	buffer.put( e ) ;
    }


    // pusher thread's body
    public void run( )
    {
	while ( true )
	    {

		// retrieve item from internal buffer
		Element item = buffer.take()  ;

		// pusher thread `pushes' data to consumer

		consumer.put( item ) ;

	    }

    }

}
// RelayBufferSystem.java


class RelayBufferSystem
{

  private RelayBuffer relaybuffer ;


  private Stage producer ;

  private PutStage consumer ;


  public RelayBufferSystem( )
  {
      // construct system in right to left order

      // first create consumer
      consumer = new RB_Consumer() ;

      // pass reference to consumer to relaybuffer
      relaybuffer = new RelayBuffer( consumer ) ;

      // pass reference to relaybuffer to producer

      producer = new RB_Producer( relaybuffer ) ;

      // start threads
      producer.start() ;

      consumer.start() ;
   }


}
// Stage.java

class Stage extends Thread
{

    private final String Descriptor = "Flow Design Pattern: Stage" ;

    public Stage( )
    {
	// Thread constructor
	super() ;

    }

    public String getDescriptor()
    {

	return Descriptor ;
    }

}
// Take.java

public interface Take
{

    abstract public Element take( ) ;

}
abstract public class TakeStage extends    Stage
		                implements Take
{


    public TakeStage( )
    {
	// Stage constructor
	super() ;

    }
}

Sorry for the long code listing, hopefully you can help me find out why the consumer is printing out null.

Thanks

That's way too much code to try to analyse just for fun.
Start to load you code with lots of System.out.println statements working backwards from the point where you get the unexpected null output - sooner or later you will find the point where it all goes wrong and then the solution will probably be obvious.

RB_Consumer is not waiting for message (from method public synchronized void put(Element item)). Prints immediately null value of item. Try set as default field private Element item = new Element("too fast"); to see. To simply trace the flow, put before all put methods line for example

public synchronized void put(Element e) {
        System.out.println("5.RelayBuffer put: " + e);
        buffer.put(e);
    }

Edited 5 Years Ago by quuba: n/a

This article has been dead for over six months. Start a new discussion instead.