Update to development state dating 20170421.

2 files changed, 734 insertions, 0 deletions

diff --git a/lib/Arctica/Serivces/Audio/Streamer/ClientGST.pm b/lib/Arctica/Serivces/Audio/Streamer/ClientGST.pm
new file mode 100644
index 0000000..969f547
--- /dev/null
+++ b/lib/Arctica/Serivces/Audio/Streamer/ClientGST.pm
@@ -0,0 +1,303 @@
+################################################################################
+#          _____ _
+#         |_   _| |_  ___
+#           | | | ' \/ -_)
+#           |_| |_||_\___|
+#                   _   _             ____            _           _
+#    / \   _ __ ___| |_(_) ___ __ _  |  _ \ _ __ ___ (_) ___  ___| |_
+#   / _ \ | '__/ __| __| |/ __/ _` | | |_) | '__/ _ \| |/ _ \/ __| __|
+#  / ___ \| | | (__| |_| | (_| (_| | |  __/| | | (_) | |  __/ (__| |_
+# /_/   \_\_|  \___|\__|_|\___\__,_| |_|   |_|  \___// |\___|\___|\__|
+#                                                  |__/
+#          The Arctica Modular Remote Computing Framework
+#
+################################################################################
+#
+# Copyright (C) 2015-2016 The Arctica Project
+# http://arctica-project.org/
+#
+# This code is dual licensed: strictly GPL-2 or AGPL-3+
+#
+# GPL-2
+# -----
+# 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.
+#
+# AGPL-3+
+# -------
+# This programm is free software; you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation; either version 3 of the License, or
+# (at your option) any later version.
+#
+# This programm 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 Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero 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.
+#
+# Copyright (C) 2015-2016 Guangzhou Nianguan Electronics Technology Co.Ltd.
+#                         <opensource@gznianguan.com>
+# Copyright (C) 2015-2016 Mike Gabriel <mike.gabriel@das-netzwerkteam.de>
+#
+################################################################################
+package Arctica::Services::Audio::Client::ClientGST;
+use strict;
+use Exporter qw(import);
+use Arctica::Core::BugOUT::Basics qw( BugOUT );
+use Arctica::Core::Mother::Forker;
+use Time::HiRes qw( usleep );
+use Data::Dumper;# Remove this before release! (unless we're still dependant)
+
+# Be very selective about what (if any) gets exported by default:
+our @EXPORT = qw( );
+# And be mindfull of what we lett the caller request here too:
+our @EXPORT_OK = qw( );
+
+my $arctica_core_object;
+
+sub new {
+	BugOUT(9,"ClientGST new->ENTER");
+	my $class_name = $_[0];# Be EXPLICIT!! DON'T SHIFT OR "@_";
+	$arctica_core_object = $_[1];
+#	my $JBUS_Server = $_[2];
+	my $self = {
+		isArctica => 1, # Declare that this is a Arctica "something"
+		aobject_name => "ClientGST",
+		default_out_rate => 64,
+		default_in_rate => 32,
+	};
+
+	bless($self, $class_name);
+
+	$arctica_core_object->{'aobj'}{'AudioClient'}{'ClientGST'} = \$self;
+
+	BugOUT(9,"ClientGST new->DONE");
+	return $self;
+}
+
+
+sub start_output {
+	BugOUT(9,"ClientGST start_output->ENTER");
+	my $self = $_[0];
+	my $idnum = $_[1];
+	my $run_on_ready = $_[2];
+	if ($idnum =~ /^(\d{1,3})$/) {
+		$idnum = $1;
+	}  else {
+		$idnum = "";
+		die;
+	}
+	my ($dsoc_type,$dsoc_port) = $self->get_device_port_and_soc_type("output",$idnum);
+
+	if ($dsoc_type and $dsoc_port) {
+		$self->{'vdev'}{'output'}{$idnum}{'running'} = 1;
+		$self->{'vdev'}{'output'}{$idnum}{'gst_thread'} = Arctica::Core::Mother::Forker->new($arctica_core_object,{
+			child_name	=>	'arctica-gst',
+			fork_style	=>	'interactive_pty',
+			handle_stdeoc	=>	sub {
+				if ($_[0] =~ /^status:sink_ready:(\d{1,3}):$/) {
+					$run_on_ready->($1);
+				}
+			},
+			return_stdin	=>	1,
+			exec_hold	=>	0,
+			exec_path	=>	"/audiotest/bin/launch_client_ThreadGST",# FIXME LOAD FULL PATH FROM CFG OR SOMETHING LIKE THAT
+			exec_cl_argv	=>	[
+							"-snk=$dsoc_type",
+							"-port=$dsoc_port",
+							"-clientside=pulseaudio",
+							"-idnum=$idnum",
+						],
+		});
+
+	} else {
+		BugOUT(1,"ClientGST start_output port and socket type not set?!! WTF?! ($dsoc_type,$dsoc_port)");
+	}
+	BugOUT(9,"ClientGST start_output->DONE");
+}
+
+
+
+sub stop_output {
+	BugOUT(9,"ClientGST stop_output->ENTER");
+	my $self = $_[0];
+	my $idnum = $_[1];
+	if ($self->{'vdev'}{'output'}{$idnum}{'gst_thread'}) {
+		$self->{'vdev'}{'output'}{$idnum}{'running'} = 0;
+		BugOUT(9,"Stop GST thread: output $idnum");
+		$self->{'vdev'}{'output'}{$idnum}{'gst_thread'}->send("cmd:stop:");
+	} else {
+		BugOUT(1,"Stop GST thread: FAIL (output $idnum)");
+	}
+	BugOUT(9,"ClientGST stop_output->DONE");
+}
+
+sub start_input {
+	BugOUT(9,"ClientGST start_input->ENTER");
+	my $self = $_[0];
+	my $idnum = $_[1];
+	if ($idnum =~ /^(\d{1,3})$/) {
+		$idnum = $1;
+	}  else {
+		$idnum = "";
+		die;
+	}
+	my $bitrate = $_[2];
+
+	if ($bitrate =~ /^(\d{1,3})$/) {
+		$bitrate = $1;
+	} else {
+		$bitrate = 0;
+	}
+
+	my ($dsoc_type,$dsoc_port) = $self->get_device_port_and_soc_type("input",$idnum);
+
+	if ($dsoc_type and $dsoc_port) {
+		$self->{'vdev'}{'input'}{$idnum}{'running'} = 1;
+		$self->{'vdev'}{'input'}{$idnum}{'gst_thread'} = Arctica::Core::Mother::Forker->new($arctica_core_object,{
+			child_name	=>	'arctica-gst',
+			fork_style	=>	'interactive_pty',
+			handle_stdeoc	=>	sub {return 1;},
+			return_stdin	=>	1,
+			exec_hold	=>	0,
+			exec_path	=>	"/audiotest/bin/launch_client_ThreadGST",# FIXME LOAD FULL PATH FROM CFG OR SOMETHING LIKE THAT
+			exec_cl_argv	=>	[
+							"-src=$dsoc_type",
+							"-port=$dsoc_port",
+							"-start_bitrate=$bitrate",
+							"-clientside=pulseaudio",
+							"-idnum=$idnum",
+						],
+		});
+	} else {
+		BugOUT(1,"ClientGST start_output port and socket type not set?!! WTF?! ($dsoc_type,$dsoc_port)");
+	}
+	BugOUT(9,"ClientGST start_input->DONE");
+}
+
+
+sub stop_input {
+	BugOUT(9,"ClientGST stop_input->ENTER");
+	my $self = $_[0];
+	my $idnum = $_[1];
+	if ($self->{'vdev'}{'input'}{$idnum}{'gst_thread'}) {
+		$self->{'vdev'}{'input'}{$idnum}{'running'} = 0;
+		BugOUT(9,"Stop GST thread: input $idnum");
+		$self->{'vdev'}{'input'}{$idnum}{'gst_thread'}->send("cmd:stop:");
+	} else {
+		BugOUT(1,"Stop GST thread: FAIL (input $idnum)");
+	}
+	BugOUT(9,"ClientGST stop_input->DONE");
+}
+
+
+sub set_bitrate {
+	BugOUT(9,"ClientGST set_bitrate->ENTER");
+	my $self = $_[0];
+	my $new_input_rate = $_[1];
+
+	if ($new_input_rate > 0) {
+			foreach my $idnum (keys %{$self->{'vdev'}{'input'}}) {
+				if ($self->{'vdev'}{'input'}{$idnum}{'running'}) {
+					if ($self->{'vdev'}{'input'}{$idnum}{'gst_thread'}) {
+						$self->{'vdev'}{'input'}{$idnum}{'gst_thread'}->send("set:bitrate:$new_input_rate");
+					}
+				}
+			}
+			BugOUT(9,"Input bitrate set to $new_input_rate");
+	} else {
+		BugOUT(9,"Input bitrate is unchanged...");
+
+	}
+	BugOUT(9,"ClientGST set_bitrate->DONE");
+}
+
+
+
+sub set_device_socket_type  {
+	BugOUT(9,"ClientGST set_device_socket_type->ENTER");
+	my $self = $_[0];
+
+	if (($_[1] eq "tcp") or ($_[1] eq "unixs")) {
+		$self->{'_settings'}{'socket_type'} = $_[1];
+		$self->{'_settings'}{'com_style'} = "stream";
+		BugOUT(9,"set_device_socket_type: socket type set to $_[1]/stream");
+	} elsif (($_[1] eq "udp") or ($_[1] eq "unixd")) {
+		$self->{'_settings'}{'socket_type'} = $_[1];
+		$self->{'_settings'}{'com_style'} = "datagram";
+		BugOUT(9,"set_device_socket_type: socket type set to $_[1]/datagram");
+	} else {
+		BugOUT(0,"set_device_socket_type: '$_[1]' is not a valid socket_type");
+	}
+
+	BugOUT(9,"ClientGST set_device_socket_type->DONE");
+}
+
+
+sub set_device_gst_port {
+	BugOUT(9,"ClientGST set_device_gst_port->ENTER");
+	my $self = $_[0];
+	my $device = $_[1];
+	my $port = $_[2];
+
+	if ($device =~ /^o(\d{1,})/) {
+		$self->{'vdev'}{'output'}{$1}{'port'} = $port;
+		BugOUT(9,"set_device_gst_port: output:$1:$port");
+	} elsif ($device =~ /^i(\d{1,})/) {
+		$self->{'vdev'}{'input'}{$1}{'port'} = $port;
+		BugOUT(9,"set_device_gst_port: input:$1:$port");
+	} else {
+		BugOUT(2,"set_device_gst_port: Failed to set device '$device' port to '$port'");
+	}
+
+	BugOUT(9,"ClientGST set_device_gst_port->DONE");
+}
+
+
+sub get_device_port_and_soc_type {
+	my $self = $_[0];
+	my $type = $_[1];
+	my $idnum = $_[2];
+
+	my $soc_type = $self->{'_settings'}{'socket_type'};
+	if ($soc_type =~ /^([a-z]{3,6})$/) {
+		$soc_type = $1;
+	} else {
+		$soc_type = 0;
+	}
+
+	if (($soc_type eq "tcp") or ($soc_type eq "udp")) {
+		if ($self->{'vdev'}{$type}{$idnum}{'port'}) {
+			my $port = $self->{'vdev'}{$type}{$idnum}{'port'};
+			if ($port =~ /^(\d{2,5})$/) {
+				$port = $1;
+			} else {
+				$port = 0;
+			}
+
+			if (($soc_type ne 0) and ($port ne 0)) {
+				return ($soc_type,$port);
+			}
+		}
+	}
+}
+
+
+1;
diff --git a/lib/Arctica/Serivces/Audio/Streamer/ClientGST/ThreadGST_client.pm b/lib/Arctica/Serivces/Audio/Streamer/ClientGST/ThreadGST_client.pm
new file mode 100644
index 0000000..4c9964a
--- /dev/null
+++ b/lib/Arctica/Serivces/Audio/Streamer/ClientGST/ThreadGST_client.pm
@@ -0,0 +1,431 @@
+################################################################################
+#          _____ _
+#         |_   _| |_  ___
+#           | | | ' \/ -_)
+#           |_| |_||_\___|
+#                   _   _             ____            _           _
+#    / \   _ __ ___| |_(_) ___ __ _  |  _ \ _ __ ___ (_) ___  ___| |_
+#   / _ \ | '__/ __| __| |/ __/ _` | | |_) | '__/ _ \| |/ _ \/ __| __|
+#  / ___ \| | | (__| |_| | (_| (_| | |  __/| | | (_) | |  __/ (__| |_
+# /_/   \_\_|  \___|\__|_|\___\__,_| |_|   |_|  \___// |\___|\___|\__|
+#                                                  |__/
+#          The Arctica Modular Remote Computing Framework
+#
+################################################################################
+#
+# Copyright (C) 2015-2016 The Arctica Project
+# http://arctica-project.org/
+#
+# This code is dual licensed: strictly GPL-2 or AGPL-3+
+#
+# GPL-2
+# -----
+# 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.
+#
+# AGPL-3+
+# -------
+# This programm is free software; you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation; either version 3 of the License, or
+# (at your option) any later version.
+#
+# This programm 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 Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero 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.
+#
+# Copyright (C) 2015-2016 Guangzhou Nianguan Electronics Technology Co.Ltd.
+#                         <opensource@gznianguan.com>
+# Copyright (C) 2015-2016 Mike Gabriel <mike.gabriel@das-netzwerkteam.de>
+#
+################################################################################
+package Arctica::Services::Audio::Client::ThreadGST_client;
+use strict;
+use Exporter qw(import);
+use Arctica::Core::BugOUT::Basics qw( BugOUT );
+#use Arctica::Core::Mother::Forker;
+use IO::Handle;
+#use Time::HiRes qw( usleep );
+use GStreamer1;
+use Data::Dumper;# Remove this before release! (unless we're still dependant)
+
+# Be very selective about what (if any) gets exported by default:
+our @EXPORT = qw( );
+# And be mindfull of what we lett the caller request here too:
+our @EXPORT_OK = qw( );
+
+my $ACO;
+
+GStreamer1::init([ $0 ]);# Initiate GST
+
+
+sub new {
+	BugOUT(9,"ThreadGST new->ENTER");
+	my $class_name = $_[0];# Be EXPLICIT!! DON'T SHIFT OR "@_";
+	$ACO = $_[1];
+
+	my $self = {
+		isArctica => 1, # Declare that this is a Arctica "something"
+		aobject_name => "ThreadGST",
+		_default_out_rate => 64,# FIXME Look for defaults in a config file somewhere?
+		_default_in_rate => 32,# FIXME	(We have a cfg file handler module somewhere.... Use it soon?)
+		_default_pasuspender_fpath => "/usr/bin/pasuspender",# FIXME  ^^^^^^^^^^
+	};
+	bless($self, $class_name);
+
+	if ($_[2]) {# FIXME!!!! We got some fancypants module somewhere that handles this.... switch to using that one, some day...!?
+		foreach my $arg (@{$_[2]}) {
+			BugOUT(8,"ARGY:\t$arg\t:ARG\n");
+			if (($arg =~ /^\-(src)\=([a-z]{3,5})/) or ($arg =~ /^\-(snk)\=([a-z]{3,5})/) ){
+	#			print "1:\t$1\n2:\t$2\n";
+				if ($1 eq "src") {
+					$self->_set_argv("src_or_snk","src");
+				} elsif ($1 eq "snk") {
+					$self->_set_argv("src_or_snk","snk");
+				} else {
+					BugOUT(9,"Not a server nor a client... what then...?");
+				}
+				if (($2 eq "tcp") or ($2 eq "unixs")) {
+					$self->_set_argv("socket_type",$2);
+					$self->_set_argv("com_style","stream");
+				} elsif (($2 eq "udp") or ($2 eq "unixd")) {
+					$self->_set_argv("socket_type",$2);
+					$self->_set_argv("com_style","datagram");
+				} else {
+					BugOUT(0,"We can't be a '$2' $self->{'s_or_c'}");
+				}
+
+			} elsif ($arg =~ /^\-oo_([a-z]{2,10})\=([a-z0-9]*)/) {
+				$self->_set_argv("opus_$1",$2);
+			} elsif ($arg =~ /^\-port\=([a-z0-9]*)/) {
+				$self->_set_argv("port_or_unix-socket",$1);
+			} elsif ($arg =~ /^\-pa_device_name\=([a-zA-Z0-9\.\_\-]*)/) {
+				BugOUT(8,"pa_device_name", $1);
+				$self->_set_argv("pa_device_name", $1);
+			} elsif ($arg =~ /^\-clientside\=([a-z]*)/) {
+				if ($1 eq "pulseaudio") {
+					$self->_set_argv("clientside","pulseaudio");
+				} else {
+					if ($1 ne "autoaudio") {
+						BugOUT(1,"-clientside=$1 ? wtf? So we're going to try to use autoaudio...");
+					}
+					$self->_set_argv("clientside","autoaudio");
+				}
+
+			} elsif ($arg =~ /^\-wait\=1/) {
+				$self->_set_argv("wait",1);
+			} elsif ($arg =~ /^\-start_bitrate\=(\d{1,3})/) {
+				$self->_set_argv("bitrate",$1);
+			} elsif ($arg =~ /^\-idnum\=(\d{1,3})/) {
+				$self->_set_argv("idnum",$1);
+			}
+
+
+		}
+
+	} else {
+		BugOUT(0,"NO ARGS?");
+	}
+
+	$ACO->{'aobj'}{'ThreadGST'} = \$self;
+
+	BugOUT(9,"ThreadGST new->DONE");
+	return $self;
+}
+
+
+sub ch_options {
+	my $self = $_[0];
+	my $o_name = $_[1];
+	my $o_value = $_[2];
+	BugOUT(8,"CH_OPTION:$o_name:$o_value");
+	if ($o_name eq "bitrate") {
+		if ($o_value =~ /^(\d{1,})$/) {
+			$self->_tune_gstpipe($1);
+		}
+	} else {
+		BugOUT(2,"WTF '$o_name'?");
+	}
+}
+
+sub _set_argv {
+	my $self = $_[0];
+	my $arg_name = $_[1];
+	my $arg_value = $_[2];
+	# FIXME DO A BUNCH OF SANETIZING HERE?!
+	$self->{'_argv'}{$arg_name} = $arg_value;
+}
+
+sub _get_argv {
+	my $self = $_[0];
+	my $arg_name = $_[1];
+	if ($self->{'_argv'}{$arg_name}) {
+		return $self->{'_argv'}{$arg_name};
+	} else {
+		return 0;
+	}
+}
+
+
+sub _tune_gstpipe {
+	my $self = $_[0];
+	my $bitrate = $_[1];
+
+	if ($self->{'main'}{'elements'}{'opusenc'}) {
+		my $opusenc_element = $self->{'main'}{'elements'}{'opusenc'};
+		$bitrate =~ s/\D//g;
+		if ($bitrate  > 384) {
+			$bitrate = 384;
+		} elsif ($bitrate  < 4) {
+			$bitrate = 4;
+		}
+		$opusenc_element->set("bitrate" => ($bitrate * 1000));
+		BugOUT(1,"\t\tTWEAK IT:\t$bitrate");
+		$opusenc_element->set("inband-fec" => 1);
+		$opusenc_element->set("max-payload-size" => 200);
+		$opusenc_element->set("frame-size" => 20);
+		#FIXME INSERT MORE COMPLEX PERFORMANCE TUNING MATRIX STUFF HERE!
+	}
+}
+
+sub _start_gstsrc {
+	my $self = $_[0];
+	$self->{'main'}{'pipeline'} = GStreamer1::Pipeline->new('pipeline');
+
+	$self->{'main'}{'elements'}{'queue1'} = GStreamer1::ElementFactory::make( queue => 'queue1' );
+	$self->{'main'}{'elements'}{'queue1'}->set("silent" => 0);
+	$self->{'main'}{'elements'}{'queue1'}->set("leaky" => "downstream");
+	$self->{'main'}{'elements'}{'queue1'}->set("max-size-time" => "30000000");
+	$self->{'main'}{'pipeline'}->add($self->{'main'}{'elements'}{'queue1'});
+
+	# This is where you'd want to add detailed support for client side stuff.....   though autoaudio(sink/src) should work fine....
+	if ($self->_get_argv("clientside") eq "pulseaudio") {
+		$self->{'main'}{'elements'}{'pasrc'} = GStreamer1::ElementFactory::make( pulsesrc => 'pasrc' );
+
+		if ($self->_get_argv("pa_device_name")) {
+			$self->{'main'}{'elements'}{'pasrc'}->set('device' => $self->_get_argv("pa_device_name"));# 'arctica.output0.monitor'
+		}
+
+		$self->{'main'}{'elements'}{'pasrc'}->set('client-name' => 'Arctica Audio Services');
+
+		$self->{'main'}{'pipeline'}->add($self->{'main'}{'elements'}{'pasrc'});
+		$self->{'main'}{'elements'}{'pasrc'}->link($self->{'main'}{'elements'}{'queue1'});
+
+	} else {
+		$self->{'main'}{'elements'}{'autoaudiosrc'} = GStreamer1::ElementFactory::make( autoaudiosrc => 'autoaudiosrc' );
+		$self->{'main'}{'elements'}{'autoaudio_convert'} = GStreamer1::ElementFactory::make( audioconvert => 'autoaudio_convert' );# cause we really don't know what the src may be
+		$self->{'main'}{'elements'}{'autoaudio_resample'} = GStreamer1::ElementFactory::make( audioresample => 'autoaudio_resample' );# cause we really don't know what the src may be
+		$self->{'main'}{'pipeline'}->add($self->{'main'}{'elements'}{'autoaudiosrc'});
+		$self->{'main'}{'pipeline'}->add($self->{'main'}{'elements'}{'autoaudio_convert'});
+		$self->{'main'}{'pipeline'}->add($self->{'main'}{'elements'}{'autoaudio_resample'});
+		$self->{'main'}{'elements'}{'autoaudiosrc'}->link($self->{'main'}{'elements'}{'autoaudio_convert'});
+		$self->{'main'}{'elements'}{'autoaudio_convert'}->link($self->{'main'}{'elements'}{'autoaudio_resample'});
+		$self->{'main'}{'elements'}{'autoaudio_resample'}->link($self->{'main'}{'elements'}{'queue1'});
+	}
+
+
+	$self->{'main'}{'elements'}{'opusenc'} = GStreamer1::ElementFactory::make( opusenc  => 'opusenc' );
+	$self->_tune_gstpipe($self->_get_argv("bitrate"));
+
+	$self->{'main'}{'pipeline'}->add($self->{'main'}{'elements'}{'opusenc'});
+	$self->{'main'}{'elements'}{'queue1'}->link($self->{'main'}{'elements'}{'opusenc'});
+
+	if ($self->_get_argv("com_style") eq "datagram")  {
+
+		$self->{'main'}{'elements'}{'rtpopuspay'} = GStreamer1::ElementFactory::make( rtpopuspay => 'rtpopuspay' );
+		$self->{'main'}{'pipeline'}->add($self->{'main'}{'elements'}{'rtpopuspay'});
+		$self->{'main'}{'elements'}{'opusenc'}->link($self->{'main'}{'elements'}{'rtpopuspay'});
+
+		if ($self->_get_argv("socket_type") eq "udp") {
+
+			$self->{'main'}{'elements'}{'udpsink'} = GStreamer1::ElementFactory::make( udpsink => 'udpsink' );
+			$self->{'main'}{'pipeline'}->add($self->{'main'}{'elements'}{'udpsink'});
+			$self->{'main'}{'elements'}{'rtpopuspay'}->link($self->{'main'}{'elements'}{'udpsink'});
+
+			$self->{'main'}{'elements'}{'udpsink'}->set('port' =>  $self->_get_argv("port_or_unix-socket"));
+			$self->{'main'}{'elements'}{'udpsink'}->set('host' =>  'localhost');
+
+		} elsif ($self->_get_argv("socket_type") eq "unixd") {
+			BugOUT(0,"NOT YET IMPLEMENTED!");
+		} else {
+			BugOUT(0,"This should never happen!");
+		}
+
+	} elsif ($self->_get_argv("com_style") eq "stream")  {
+
+		$self->{'main'}{'elements'}{'gdppay'} = GStreamer1::ElementFactory::make( gdppay => 'gdppay' );
+		$self->{'main'}{'pipeline'}->add($self->{'main'}{'elements'}{'gdppay'});
+		$self->{'main'}{'elements'}{'opusenc'}->link($self->{'main'}{'elements'}{'gdppay'});
+#
+		$self->{'main'}{'elements'}{'queue2'} = GStreamer1::ElementFactory::make( queue => 'queue2' );
+		$self->{'main'}{'elements'}{'queue2'}->set("silent" => 0);
+		$self->{'main'}{'elements'}{'queue2'}->set("leaky" => "downstream");
+		$self->{'main'}{'elements'}{'queue2'}->set("max-size-time" => "30000000");
+		$self->{'main'}{'pipeline'}->add($self->{'main'}{'elements'}{'queue2'});
+
+		$self->{'main'}{'elements'}{'gdppay'}->link($self->{'main'}{'elements'}{'queue2'});
+
+		if ($self->_get_argv("socket_type")  eq "tcp") {
+			$self->{'main'}{'elements'}{'tcpclientsink'} = GStreamer1::ElementFactory::make( tcpclientsink => 'tcpclientsink' );
+			$self->{'main'}{'elements'}{'tcpclientsink'}->set('port' =>  $self->_get_argv("port_or_unix-socket"));
+			$self->{'main'}{'elements'}{'tcpclientsink'}->set('host' =>  'localhost');
+			$self->{'main'}{'pipeline'}->add($self->{'main'}{'elements'}{'tcpclientsink'});
+			$self->{'main'}{'elements'}{'queue2'}->link($self->{'main'}{'elements'}{'tcpclientsink'});
+
+		} elsif ($self->_get_argv("socket_type")  eq "unixs") {
+			BugOUT(0,"NOT YET IMPLEMENTED!");
+		} else {
+			BugOUT(0,"This should never happen!");
+		}
+
+	} else {
+		BugOUT(0,"com_style missing? WTF this should never happen at this point!");
+	}
+
+	$self->{'main'}{'pipeline'}->set_state("playing");
+	BugOUT(8,"Main pipeline initiated");
+
+
+}
+
+
+
+sub _start_gstsnk {
+	BugOUT(9,"ThreadGST _start_gstsnk->ENTER");
+	my $self = $_[0];
+
+	$self->{'main'}{'pipeline'} = GStreamer1::Pipeline->new('pipeline');
+
+	$self->{'main'}{'elements'}{'opusdec'} = GStreamer1::ElementFactory::make( opusdec => 'opusdec' );
+	$self->{'main'}{'pipeline'}->add($self->{'main'}{'elements'}{'opusdec'});
+
+	if ($self->_get_argv("com_style") eq "datagram")  {
+		if ($self->_get_argv("socket_type") eq "udp") {
+
+			$self->{'main'}{'elements'}{'udpsrc'} = GStreamer1::ElementFactory::make( udpsrc => 'udpsrc' );
+			$self->{'main'}{'elements'}{'udpsrc'}->set('port' =>  $self->_get_argv("port_or_unix-socket"));
+			$self->{'main'}{'elements'}{'udpsrc'}->set( caps => GStreamer1::Caps::Simple->new(
+										'application/x-rtp',
+										'media' =>  'Glib::String' => 'audio',
+										'clock-rate' =>  'Glib::Int' => 48000,
+										'encoding-name' =>  'Glib::String' => 'X-GST-OPUS-DRAFT-SPITTKA-00'));
+
+			$self->{'main'}{'pipeline'}->add($self->{'main'}{'elements'}{'udpsrc'});
+
+			$self->{'main'}{'elements'}{'rtpopusdepay'} = GStreamer1::ElementFactory::make( rtpopusdepay => 'rtpopusdepay' );
+			$self->{'main'}{'pipeline'}->add($self->{'main'}{'elements'}{'rtpopusdepay'});
+
+			$self->{'main'}{'elements'}{'udpsrc'}->link($self->{'main'}{'elements'}{'rtpopusdepay'});
+			$self->{'main'}{'elements'}{'rtpopusdepay'}->link($self->{'main'}{'elements'}{'opusdec'});
+
+		} elsif ($self->_get_argv("socket_type") eq "unixd") {
+			BugOUT(0,"NOT YET IMPLEMENTED!");
+		} else {
+			BugOUT(0,"This should never happen!");
+		}
+	} elsif ($self->_get_argv("com_style") eq "stream")  {
+		if ($self->_get_argv("socket_type")  eq "tcp") {
+
+			$self->{'main'}{'elements'}{'tcpserversrc'} = GStreamer1::ElementFactory::make( tcpserversrc => 'tcpserversrc' );
+			$self->{'main'}{'elements'}{'tcpserversrc'}->set('port' =>  $self->_get_argv("port_or_unix-socket"));
+			$self->{'main'}{'pipeline'}->add($self->{'main'}{'elements'}{'tcpserversrc'});
+
+			$self->{'main'}{'elements'}{'queue1'} = GStreamer1::ElementFactory::make( queue => 'queue1' );
+			$self->{'main'}{'elements'}{'queue1'}->set("silent" => 0);
+			$self->{'main'}{'elements'}{'queue1'}->set("leaky" => "downstream");
+			$self->{'main'}{'elements'}{'queue1'}->set("max-size-time" => "30000000");
+			$self->{'main'}{'pipeline'}->add($self->{'main'}{'elements'}{'queue1'});
+
+			$self->{'main'}{'elements'}{'tcpserversrc'}->link($self->{'main'}{'elements'}{'queue1'});
+
+			$self->{'main'}{'elements'}{'gdpdepay'} = GStreamer1::ElementFactory::make( gdpdepay => 'gdpdepay' );
+			$self->{'main'}{'pipeline'}->add($self->{'main'}{'elements'}{'gdpdepay'});
+
+			$self->{'main'}{'elements'}{'queue1'}->link($self->{'main'}{'elements'}{'gdpdepay'});
+			$self->{'main'}{'elements'}{'gdpdepay'}->link($self->{'main'}{'elements'}{'opusdec'});
+
+		} elsif ($self->_get_argv("socket_type")  eq "unixs") {
+			BugOUT(0,"NOT YET IMPLEMENTED!");
+		} else {
+			BugOUT(0,"This should never happen!");
+		}
+	}
+
+
+	if ($self->_get_argv("clientside") eq "pulseaudio") {
+
+		$self->{'main'}{'elements'}{'pasink'} = GStreamer1::ElementFactory::make( pulsesink => 'pasink' );
+
+		if ($self->_get_argv("pa_device_name")) {
+			$self->{'main'}{'elements'}{'pasink'}->set('device' => $self->_get_argv("pa_device_name"));
+		}
+
+		$self->{'main'}{'elements'}{'pasink'}->set('client-name' => 'Arctica Audio Services');
+
+		$self->{'main'}{'pipeline'}->add($self->{'main'}{'elements'}{'pasink'});
+		$self->{'main'}{'elements'}{'opusdec'}->link($self->{'main'}{'elements'}{'pasink'});
+
+	} else {
+# 		Since we really don't know what the sink may be:
+		$self->{'main'}{'elements'}{'autoaudio_convert'} = GStreamer1::ElementFactory::make( audioconvert => 'autoaudio_convert' );
+		$self->{'main'}{'elements'}{'autoaudio_resample'} = GStreamer1::ElementFactory::make( audioresample => 'autoaudio_resample' );
+		$self->{'main'}{'elements'}{'autoaudiosink'} = GStreamer1::ElementFactory::make( autoaudiosink => 'autoaudiosink' );
+
+		$self->{'main'}{'pipeline'}->add($self->{'main'}{'elements'}{'autoaudio_convert'});
+		$self->{'main'}{'pipeline'}->add($self->{'main'}{'elements'}{'autoaudio_resample'});
+		$self->{'main'}{'pipeline'}->add($self->{'main'}{'elements'}{'autoaudiosink'});
+
+		$self->{'main'}{'elements'}{'opusdec'}->link($self->{'main'}{'elements'}{'autoaudio_convert'});
+		$self->{'main'}{'elements'}{'autoaudio_convert'}->link($self->{'main'}{'elements'}{'autoaudio_resample'});
+		$self->{'main'}{'elements'}{'autoaudio_resample'}->link($self->{'main'}{'elements'}{'autoaudiosink'});
+
+	}
+
+
+
+	$self->{'main'}{'pipeline'}->set_state("playing");
+	BugOUT(8,"Main sink pipeline initiated");
+	print "status:sink_ready:",$self->_get_argv("idnum"),":\n";
+	BugOUT(9,"ThreadGST _start_gstsnk->DONE");
+}
+
+sub start {
+	my $self = $_[0];
+	if ($self->_get_argv("src_or_snk") eq "src") {
+		$self->_start_gstsrc;
+	} elsif ($self->_get_argv("src_or_snk") eq "snk") {
+		$self->_start_gstsnk;
+	} else {
+		BugOUT(0,"WTF? (src_or_snk!!)")
+	}
+}
+
+sub terminate {
+	my $self = $_[0];
+	if ($self->{'main'}{'pipeline'}) {
+		$self->{'main'}{'pipeline'}->set_state("paused");
+		$self->{'main'}{'pipeline'}->set_state("null");
+	}
+	# FIXME! Do something else too??
+}
+
+1;
+