Revision: 2a9024a71e12
Author: unknown <yulian@Venera.(none)>
Date: Fri Jan 4 03:50:24 2013
Log: OOB Stage 1 - Compound components
http://code.google.com/p/mediaserver/source/detail?r=2a9024a71e12
Added:
/component/src/main/java/org/mobicents/media/server/component/oob/OOBComponent.java
/component/src/main/java/org/mobicents/media/server/component/oob/OOBInput.java
/component/src/main/java/org/mobicents/media/server/component/oob/OOBMixer.java
/component/src/main/java/org/mobicents/media/server/component/oob/OOBOutput.java
/component/src/main/java/org/mobicents/media/server/component/oob/OOBSplitter.java
=======================================
--- /dev/null
+++
/component/src/main/java/org/mobicents/media/server/component/oob/OOBComponent.java
Fri Jan 4 03:50:24 2013
@@ -0,0 +1,133 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2011, Red Hat, Inc. and individual contributors
+ * by the @authors tag. See the copyright.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software 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
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+
+package org.mobicents.media.server.component.oob;
+
+import java.util.Iterator;
+import org.mobicents.media.server.scheduler.IntConcurrentLinkedList;
+import org.mobicents.media.server.spi.format.AudioFormat;
+import org.mobicents.media.server.spi.format.Format;
+import org.mobicents.media.server.spi.format.FormatFactory;
+import org.mobicents.media.server.spi.memory.Frame;
+import org.mobicents.media.server.spi.memory.Memory;
+
+/**
+ * Implements compound components used by mixer and splitter.
+ *
+ * @author Yulian Oifa
+ */
+public class OOBComponent {
+ //the format of the output stream.
+ private final static AudioFormat dtmf =
FormatFactory.createAudioFormat("telephone-event", 8000);
+
+ private IntConcurrentLinkedList<OOBInput> inputs = new
IntConcurrentLinkedList();
+ private IntConcurrentLinkedList<OOBOutput> outputs = new
IntConcurrentLinkedList();
+
+ protected Boolean shouldRead=false;
+ protected Boolean shouldWrite=false;
+
+ private Frame frame;
+
+ private int componentId;
+ /**
+ * Creates new instance with default name.
+ */
+ public OOBComponent(int componentId) {
+ this.componentId=componentId;
+ }
+
+ public int getComponentId()
+ {
+ return componentId;
+ }
+
+ public void updateMode(Boolean shouldRead,Boolean shouldWrite)
+ {
+ this.shouldRead=shouldRead;
+ this.shouldWrite=shouldWrite;
+ }
+
+ public void addInput(OOBInput input) {
+ inputs.offer(input,input.getInputId());
+ }
+
+ public void addOutput(OOBOutput output) {
+ outputs.offer(output,output.getOutputId());
+ }
+
+ public void remove(OOBInput input)
+ {
+ inputs.remove(input.getInputId());
+ }
+
+ public void remove(OOBOutput output)
+ {
+ outputs.remove(output.getOutputId());
+ }
+
+ public void perform()
+ {
+ frame=null;
+ Iterator<OOBInput> activeInputs=inputs.iterator();
+ while(activeInputs.hasNext())
+ {
+ OOBInput input=activeInputs.next();
+ frame=input.poll();
+ if(frame!=null)
+ break;
+ }
+ }
+
+ public Frame getData()
+ {
+ if(!this.shouldRead)
+ {
+ if(frame!=null)
+ frame.recycle();
+
+ return null;
+ }
+
+ return frame;
+ }
+
+ public void offer(Frame frame)
+ {
+ if(!this.shouldWrite)
+ {
+ frame.recycle();
+ return;
+ }
+
+ Iterator<OOBOutput> activeOutputs=outputs.iterator();
+ while(activeOutputs.hasNext())
+ {
+ OOBOutput output=activeOutputs.next();
+ if(!activeOutputs.hasNext())
+ output.offer(frame);
+ else
+ output.offer(frame.clone());
+
+ output.wakeup();
+ }
+ }
+}
=======================================
--- /dev/null
+++
/component/src/main/java/org/mobicents/media/server/component/oob/OOBInput.java
Fri Jan 4 03:50:24 2013
@@ -0,0 +1,104 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2011, Red Hat, Inc. and individual contributors
+ * by the @authors tag. See the copyright.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software 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
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+
+package org.mobicents.media.server.component.oob;
+
+import java.io.IOException;
+
+import org.mobicents.media.server.impl.AbstractSink;
+import org.mobicents.media.server.scheduler.ConcurrentLinkedList;
+import org.mobicents.media.server.spi.memory.Frame;
+import org.mobicents.media.server.spi.memory.Memory;
+
+/**
+ * Implements input for compound components
+ *
+ * @author Yulian Oifa
+ */
+public class OOBInput extends AbstractSink {
+ private int inputId;
+ private int limit=10;
+ private ConcurrentLinkedList<Frame> buffer = new
ConcurrentLinkedList();
+
+ /**
+ * Creates new stream
+ */
+ public OOBInput(int inputId) {
+ super("compound.input");
+ this.inputId=inputId;
+ }
+
+ public int getInputId()
+ {
+ return inputId;
+ }
+
+ public void activate()
+ {
+
+ }
+
+ public void deactivate()
+ {
+
+ }
+
+ @Override
+ public void onMediaTransfer(Frame frame) throws IOException {
+ if (buffer.size() >= limit)
+ buffer.poll().recycle();
+
+ buffer.offer(frame);
+ }
+
+ /**
+ * Indicates the state of the input buffer.
+ *
+ * @return true if input buffer has no frames.
+ */
+ public boolean isEmpty() {
+ return buffer.isEmpty();
+ }
+
+ /**
+ * Retrieves frame from the input buffer.
+ *
+ * @return the media frame.
+ */
+ public Frame poll() {
+ return buffer.poll();
+ }
+
+ /**
+ * Recycles input stream
+ */
+ public void recycle() {
+ while(buffer.size()>0)
+ buffer.poll().recycle();
+ }
+
+ public void resetBuffer()
+ {
+ while(buffer.size()>0)
+ buffer.poll().recycle();
+ }
+}
=======================================
--- /dev/null
+++
/component/src/main/java/org/mobicents/media/server/component/oob/OOBMixer.java
Fri Jan 4 03:50:24 2013
@@ -0,0 +1,137 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2011, Red Hat, Inc. and individual contributors
+ * by the @authors tag. See the copyright.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software 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
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+
+package org.mobicents.media.server.component.oob;
+
+import java.util.Iterator;
+
+import org.mobicents.media.server.scheduler.Scheduler;
+import org.mobicents.media.server.scheduler.Task;
+import org.mobicents.media.server.scheduler.IntConcurrentLinkedList;
+import org.mobicents.media.server.spi.format.AudioFormat;
+import org.mobicents.media.server.spi.format.Format;
+import org.mobicents.media.server.spi.format.FormatFactory;
+import org.mobicents.media.server.spi.memory.Frame;
+import org.mobicents.media.server.spi.memory.Memory;
+
+/**
+ * Implements compound oob mixer , one of core components of mms 3.0
+ *
+ * @author Yulian Oifa
+ */
+public class OOBMixer {
+ //scheduler for mixer job scheduling
+ private Scheduler scheduler;
+
+ //The pool of components
+ private IntConcurrentLinkedList<OOBComponent> components = new
IntConcurrentLinkedList();
+
+ private MixTask mixer;
+ private volatile boolean started = false;
+
+ public long mixCount = 0;
+
+ //gain value
+ private double gain = 1.0;
+
+ public OOBMixer(Scheduler scheduler) {
+ this.scheduler = scheduler;
+
+ mixer = new MixTask();
+ }
+
+ public void addComponent(OOBComponent component)
+ {
+ components.offer(component,component.getComponentId());
+ }
+
+ /**
+ * Releases unused input stream
+ *
+ * @param input the input stream previously created
+ */
+ public void release(OOBComponent component) {
+ components.remove(component.getComponentId());
+ }
+
+ public void start() {
+ mixCount = 0;
+ started = true;
+ scheduler.submit(mixer,scheduler.MIXER_MIX_QUEUE);
+ }
+
+ public void stop() {
+ started = false;
+ mixer.cancel();
+ }
+
+ private class MixTask extends Task {
+ int sourceComponent=0;
+ private Frame current;
+
+ public MixTask() {
+ super();
+ }
+
+ public int getQueueNumber()
+ {
+ return scheduler.MIXER_MIX_QUEUE;
+ }
+
+ public long perform() {
+ //summarize all
+ Iterator<OOBComponent> activeComponents=components.iterator();
+ while(activeComponents.hasNext())
+ {
+ OOBComponent component=activeComponents.next();
+ component.perform();
+ current=component.getData();
+ if(current!=null)
+ {
+ sourceComponent=component.getComponentId();
+ break;
+ }
+ }
+
+ if(current==null)
+ {
+ scheduler.submit(this,scheduler.MIXER_MIX_QUEUE);
+ mixCount++;
+ return 0;
+ }
+
+ //get data for each component
+ activeComponents=components.iterator();
+ while(activeComponents.hasNext())
+ {
+ OOBComponent component=activeComponents.next();
+ if(component.getComponentId()!=sourceComponent)
+ component.offer(current.clone());
+ }
+
+ current.recycle();
+ scheduler.submit(this,scheduler.MIXER_MIX_QUEUE);
+ mixCount++;
+ return 0;
+ }
+ }
+}
=======================================
--- /dev/null
+++
/component/src/main/java/org/mobicents/media/server/component/oob/OOBOutput.java
Fri Jan 4 03:50:24 2013
@@ -0,0 +1,99 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2011, Red Hat, Inc. and individual contributors
+ * by the @authors tag. See the copyright.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software 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
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+
+package org.mobicents.media.server.component.oob;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.mobicents.media.MediaSource;
+import org.mobicents.media.server.impl.AbstractSource;
+import org.mobicents.media.server.impl.AbstractSink;
+import org.mobicents.media.server.scheduler.Scheduler;
+import org.mobicents.media.server.scheduler.Task;
+import org.mobicents.media.server.scheduler.ConcurrentLinkedList;
+import org.mobicents.media.server.spi.memory.Frame;
+
+import org.apache.log4j.Logger;
+/**
+ * Implements output for compound components.
+ *
+ * @author Yulian Oifa
+ */
+public class OOBOutput extends AbstractSource {
+ private int outputId;
+ private ConcurrentLinkedList<Frame> buffer = new
ConcurrentLinkedList();
+
+ /**
+ * Creates new instance with default name.
+ */
+ public OOBOutput(Scheduler scheduler,int outputId) {
+ super("compound.output", scheduler,scheduler.OUTPUT_QUEUE);
+ this.outputId=outputId;
+ }
+
+ public int getOutputId()
+ {
+ return outputId;
+ }
+
+ public void join(AbstractSink sink)
+ {
+ connect(sink);
+ }
+
+ public void unjoin()
+ {
+ disconnect();
+ }
+
+ @Override
+ public Frame evolve(long timestamp) {
+ return buffer.poll();
+ }
+
+ @Override
+ public void stop() {
+ while(buffer.size()>0)
+ buffer.poll().recycle();
+
+ super.stop();
+ }
+
+ public void resetBuffer()
+ {
+ while(buffer.size()>0)
+ buffer.poll().recycle();
+ }
+
+ public void offer(Frame frame)
+ {
+ if(buffer.size()>1)
+ buffer.poll().recycle();
+
+ buffer.offer(frame);
+ }
+}
=======================================
--- /dev/null
+++
/component/src/main/java/org/mobicents/media/server/component/oob/OOBSplitter.java
Fri Jan 4 03:50:24 2013
@@ -0,0 +1,202 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2011, Red Hat, Inc. and individual contributors
+ * by the @authors tag. See the copyright.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software 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
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+
+package org.mobicents.media.server.component.oob;
+
+import java.util.Iterator;
+
+import org.mobicents.media.server.scheduler.Scheduler;
+import org.mobicents.media.server.scheduler.Task;
+import org.mobicents.media.server.scheduler.IntConcurrentLinkedList;
+import org.mobicents.media.server.spi.format.AudioFormat;
+import org.mobicents.media.server.spi.format.Format;
+import org.mobicents.media.server.spi.format.FormatFactory;
+import org.mobicents.media.server.spi.memory.Frame;
+import org.mobicents.media.server.spi.memory.Memory;
+
+/**
+ * Implements compound oob splitter , one of core components of mms 3.0
+ *
+ * @author Yulian Oifa
+ */
+public class OOBSplitter {
+ //scheduler for mixer job scheduling
+ private Scheduler scheduler;
+
+ //The pools of components
+ private IntConcurrentLinkedList<OOBComponent> insideComponents = new
IntConcurrentLinkedList();
+ private IntConcurrentLinkedList<OOBComponent> outsideComponents = new
IntConcurrentLinkedList();
+
+ private InsideMixTask insideMixer;
+ private OutsideMixTask outsideMixer;
+ private volatile boolean started = false;
+
+ protected long mixCount = 0;
+
+ public OOBSplitter(Scheduler scheduler) {
+ this.scheduler = scheduler;
+
+ insideMixer = new InsideMixTask();
+ outsideMixer = new OutsideMixTask();
+ }
+
+ public void addInsideComponent(OOBComponent component)
+ {
+ insideComponents.offer(component,component.getComponentId());
+ }
+
+ public void addOutsideComponent(OOBComponent component)
+ {
+ outsideComponents.offer(component,component.getComponentId());
+ }
+
+ /**
+ * Releases inside component
+ *
+ * @param component
+ */
+ public void releaseInsideComponent(OOBComponent component) {
+ insideComponents.remove(component.getComponentId());
+ }
+
+ /**
+ * Releases outside component
+ *
+ * @param component
+ */
+ public void releaseOutsideComponent(OOBComponent component) {
+ outsideComponents.remove(component.getComponentId());
+ }
+
+ public void start() {
+ mixCount = 0;
+ started = true;
+ scheduler.submit(insideMixer,scheduler.MIXER_MIX_QUEUE);
+ scheduler.submit(outsideMixer,scheduler.MIXER_MIX_QUEUE);
+ }
+
+ public void stop() {
+ started = false;
+ insideMixer.cancel();
+ outsideMixer.cancel();
+ }
+
+ private class InsideMixTask extends Task {
+ private Frame current;
+
+ public InsideMixTask() {
+ super();
+ }
+
+ public int getQueueNumber()
+ {
+ return scheduler.MIXER_MIX_QUEUE;
+ }
+
+ public long perform() {
+ //summarize all
+ current=null;
+ Iterator<OOBComponent>
activeComponents=insideComponents.iterator();
+ while(activeComponents.hasNext())
+ {
+ OOBComponent component=activeComponents.next();
+ component.perform();
+ current=component.getData();
+ if(current!=null)
+ break;
+ }
+
+ if(current==null)
+ {
+ scheduler.submit(this,scheduler.MIXER_MIX_QUEUE);
+ mixCount++;
+ return 0;
+ }
+
+ //get data for each component
+ activeComponents=outsideComponents.iterator();
+ while(activeComponents.hasNext())
+ {
+ OOBComponent component=activeComponents.next();
+ if(!activeComponents.hasNext())
+ component.offer(current);
+ else
+ component.offer(current.clone());
+ }
+
+ scheduler.submit(this,scheduler.MIXER_MIX_QUEUE);
+ mixCount++;
+
+ return 0;
+ }
+ }
+
+ private class OutsideMixTask extends Task {
+ private Frame current;
+
+ public OutsideMixTask() {
+ super();
+ }
+
+ public int getQueueNumber()
+ {
+ return scheduler.MIXER_MIX_QUEUE;
+ }
+
+ public long perform() {
+ //summarize all
+ current=null;
+ Iterator<OOBComponent>
activeComponents=outsideComponents.iterator();
+ while(activeComponents.hasNext())
+ {
+ OOBComponent component=activeComponents.next();
+ component.perform();
+ current=component.getData();
+ if(current!=null)
+ break;
+ }
+
+ if(current==null)
+ {
+ scheduler.submit(this,scheduler.MIXER_MIX_QUEUE);
+ mixCount++;
+ return 0;
+ }
+
+ //get data for each component
+ activeComponents=insideComponents.iterator();
+ while(activeComponents.hasNext())
+ {
+ OOBComponent component=activeComponents.next();
+ if(!activeComponents.hasNext())
+ component.offer(current);
+ else
+ component.offer(current.clone());
+ }
+
+ scheduler.submit(this,scheduler.MIXER_MIX_QUEUE);
+ mixCount++;
+
+ return 0;
+ }
+ }
+}
