WriteableByteChannelSink Update Again Relates to issue 91

[GoogleCodeExporter]

Hi Farkas

I have made some more changes. I have created a new lock class, and updated the 
classes ReadableByteChannelSrc and WriteableByteChannelSink to use it. 

The changes mean that a 'main' thread could create the above input/output 
classes, then wait on the new lock object, until someone signals completion. 
i.e. the bus callbacks or alternatively an error from somewhere. If any NIO 
error occurs within the input/output classes then it will signal the lock.

Here is some sample code, this is our main thread :

lock = new StreamLock();
outputStreamSink = new OutputStreamSink(outputstream, "outputStream");
outputStreamSink.setAutoFlushBuffer(true);
outputStreamSink.set("sync", false);
outputStreamSink.setNotifyOnError(lock);
inputStreamSrc.setNotifyOnError(lock);

pipeline.setState(State.PLAYING);

synchronized (lock) {
while (lock.isDone() == false) {
try {
lock.wait();
} catch (InterruptedException e) {
                    }
                }
            }

I have left some things out, but above is just a pointer. To active this code, 
you have to call setNotifyOnError with the lock object. Without this call, 
everything works as before.

Here is a diff of the new changes :

Index: src/org/gstreamer/io/WriteableByteChannelSink.java
===================================================================
--- src/org/gstreamer/io/WriteableByteChannelSink.java  (revision 536)
+++ src/org/gstreamer/io/WriteableByteChannelSink.java  (working copy)
@@ -24,23 +24,55 @@
 import org.gstreamer.Buffer;
 import org.gstreamer.FlowReturn;
 import org.gstreamer.elements.CustomSink;
-
+import org.gstreamer.io.StreamLock;
 /**
  *
  * @author wayne
  */
-public class WriteableByteChannelSink extends CustomSink {
+public class WriteableByteChannelSink extends CustomSink{

     private WritableByteChannel channel;
+    private boolean autoFlushBuffer = false;
+    private StreamLock lock = null;

     public WriteableByteChannelSink(final WritableByteChannel channel, String name) {
         super(WriteableByteChannelSink.class, name);
         this.channel = channel;
     }

-    @Override
+    public void setAutoFlushBuffer(boolean value){
+       this.autoFlushBuffer = value;
+    }
+    
+    public void setNotifyOnError(StreamLock lock){
+       this.lock = lock;
+    }
+    
+    private void signalError(){
+       if(null != lock){
+           lock.setDone();
+       }
+    }
+    
+    
+   @Override
     protected final FlowReturn sinkRender(Buffer buffer) throws IOException {
-        channel.write(buffer.getByteBuffer());
-        return FlowReturn.OK;
-    }   
+       
+       try{
+           channel.write(buffer.getByteBuffer());
+           return FlowReturn.OK;
+       }catch(IOException ex){
+           signalError();
+           return FlowReturn.ERROR;
+       }finally{
+            //Dispose immediate to avoid GC Delay
+            if(autoFlushBuffer){
+               if((null != buffer) && (null != buffer.getAddress())){
+                   buffer.dispose();
+               }
+            }
+       }
+    }
+    
+    
 }
Index: src/org/gstreamer/io/ReadableByteChannelSrc.java
===================================================================
--- src/org/gstreamer/io/ReadableByteChannelSrc.java    (revision 536)
+++ src/org/gstreamer/io/ReadableByteChannelSrc.java    (working copy)
@@ -38,6 +38,8 @@
     private final ReadableByteChannel channel;
     private FileChannel fileChannel;
     private long channelPosition = 0;
+    private StreamLock lock = null;
+    
     public ReadableByteChannelSrc(ReadableByteChannel src, String name) {
         super(ReadableByteChannelSrc.class, name);  
         this.channel = src;
@@ -83,6 +85,7 @@
             readFully(offset, size, buffer);
             return FlowReturn.OK;
         } catch (IOException ex) {
+           signalError();
 //            System.out.println(ex);
             return FlowReturn.UNEXPECTED;
         }        
@@ -104,6 +107,7 @@
                 segment.write();
                 return true;
             } catch (IOException ex) {
+               signalError();
                 Logger.getLogger(getClass().getName()).log(Level.SEVERE, null, ex);
                 return false;
             }
@@ -119,6 +123,7 @@
             try {
                 return fileChannel.size();
             } catch (IOException ex) {
+               signalError();
                 Logger.getLogger(ReadableByteChannelSrc.class.getName()).log(Level.SEVERE, null, ex);
                 return -1;
             }
@@ -126,4 +131,14 @@
         // We can't figure out the size of non-filechannel files
         return -1;
     }
+    
+    private void signalError(){
+       if(null != lock){
+           lock.setDone();
+       }
+    }
+    
+    public void setNotifyOnError(StreamLock lock){
+       this.lock = lock;
+    }
 }

And here is the new StreamLock class at 
gstreamer-java-orig/gstreamer-java/src/org/gstreamer/io/StreamLock.java

package org.gstreamer.io;

public class StreamLock {
        protected boolean done;

        public StreamLock(){
                done = false;
        }

        public boolean isDone() {
                return done;
        }

        public void setDone(){
                synchronized(this){
                        done = true;
                        this.notifyAll();
                }
        }

}

Original issue reported on code.google.com by ch...@crazyfool.org on 21 Dec 2011 at 5:29

[GoogleCodeExporter]

please add attachment patch not inline...

Original comment by lfar...@lfarkas.org on 21 Dec 2011 at 9:22

[GoogleCodeExporter]

Hi sorry about the delay

I will get a patch added this week, have been away from the computer over xmas.

C

Original comment by ch...@crazyfool.org on 3 Jan 2012 at 1:08

[GoogleCodeExporter]

ping.

Original comment by vitorl...@gmail.com on 4 Jan 2012 at 12:52

[GoogleCodeExporter]

Sorry for delay diff attached.

Original comment by ch...@crazyfool.org on 10 Jan 2012 at 12:00

Attachments:

• diff

[GoogleCodeExporter]

And also attached is new StreamLock.java file which above diff requires.

Original comment by ch...@crazyfool.org on 10 Jan 2012 at 12:02

Attachments:

• StreamLock.java

[GoogleCodeExporter]

what's the reason for org/gstreamer/Buffer.java patch?

Original comment by lfar...@lfarkas.org on 13 Jan 2012 at 9:13

[GoogleCodeExporter]

next time please create the patch against the svn head! i have to edit 
everything manually.
please check the current svn head.

Original comment by lfar...@lfarkas.org on 13 Jan 2012 at 9:59

• Changed state: Fixed