#!/bin/bash

# Relational pipes

# Copyright © 2020 František Kučera (Frantovo.cz, GlobalCode.info)

#

# 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 3 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, see <http://www.gnu.org/licenses/>.

# Unlike the protocol and the message format,

# these helper functions are not part of the public API.

# Thus when writing custom streamlets, it is better to copy this file

# and review its changes while upgrading to new upstream version.

send() {

	local msgVar="EXEC_MSG_$1";

	printf '%s\0' "${!msgVar}"

	shift;

	printf '%s\0' "$#"

	for param in "$@"; do printf '%s\0' "$param"; done 

}

read_nullbyte() { local IFS=; for v in "$@"; do export "$v"; read -r -d '' "$v"; done }

processMessages() {

	while read_nullbyte code length; do

		local msgVar="EXEC_MSG_$code";

		local functionName="processMessage_${!msgVar}";

		local parameters=();

		for (( i=0; i<$length; i++)); do

			read_nullbyte parameter;

			parameters+=("$parameter")

		done

		type "$functionName" &>/dev/null || functionName="processUnsupportedMessage";

		"$functionName" "${parameters[@]}";

	done

}

processUnsupportedMessage() {

	echo "Unsupported message: $*" >&2

	# TODO: send error, no debug

	# send STREAMLET_ERROR "UNSUPPORTED_MESSAGE"

}

initialize() {

	versionsSupported=();

	optionNames=();

	optionValues=();

	inputAttributeNames=();

	inputAttributeTypes=();

	outputAttributeAliases=();

	currentRelation="";

	currentFile="";

}

processMessage_VERSION_SUPPORTED() {

	versionsSupported+=("$1");

}

processMessage_WAITING_FOR_VERSION() {

	for v in "${versionsSupported[@]}"; do

		if [[ "x$v" == "x1" ]]; then

			send VERSION_ACCEPTED 1;

			return;

		fi

	done

	send STREAMLET_ERROR "INCOMPATIBLE_VERSION" "Only version 1 is supported by this streamlet."

}

processMessage_RELATION_START() {

	currentRelation="$1";

	return;

}

processMessage_INPUT_ATTRIBUTE_METADATA() {

	inputAttributeNames+=("$1");

	inputAttributeTypes+=("$2");

}

processMessage_OUTPUT_ATTRIBUTE_ALIAS() {

	outputAttributeAliases+=("$1");

}

processMessage_OPTION() {

	optionNames+=("$1");

	optionValues+=("$2");

}

processMessage_INPUT_ATTRIBUTE() {

	local index="$1";

	local value="$2";

	local isNull="$3";

	if [[ "_${inputAttributeNames[$index]}" == "_path" ]]; then

		currentFile="$value";

	fi

}

processMessage_RELATION_END() {

	# Now we should stop reading messages,

	# but the executor also closes our STDIN,

	# so the loop in processMessages() terminates.

	# Override this function to do some clean-up.

	return;

}