text:
package com.groupon.stomp;

import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Map;
import org.apache.log4j.Logger;

import com.groupon.messagebus.api.exceptions.InvalidDestinationException;
import com.groupon.messagebus.api.exceptions.MessageBusException;
import com.groupon.messagebus.api.exceptions.SendFailedException;
import com.groupon.messagebus.client.Utils;

public class StompConnection {

    // Set this to appropriate value for now set to infinity
    public static final long RECEIVE_TIMEOUT = 30000;

    private Logger log = Logger.getLogger(StompConnection.class);
    private boolean connected = false;
    private Socket stompSocket;
    private ByteArrayOutputStream inputBuffer = new ByteArrayOutputStream();

    public void open(String host, int port) throws IOException,
            UnknownHostException {
        Socket sock = new Socket(host, port);
        sock.setTcpNoDelay(true);
        open(sock);
    }

    public void open(Socket socket) {
        stompSocket = socket;
        connected = true;

    }

    public void close() throws IOException {
        if (stompSocket != null) {
            if (!stompSocket.isClosed())
                stompSocket.close();
            stompSocket = null;
            connected = false;
        }
    }

    public void sendFrame(String data) throws IOException {
        byte[] bytes = data.getBytes("UTF-8");
        OutputStream outputStream = stompSocket.getOutputStream();
        outputStream.write(bytes);
        outputStream.write(0);
        outputStream.flush();
    }

    public void sendFrame(String frame, byte[] data) throws IOException {
        byte[] bytes = frame.getBytes("UTF-8");
        OutputStream outputStream = stompSocket.getOutputStream();
        outputStream.write(bytes);
        outputStream.write(data);
        outputStream.write(0);
        outputStream.flush();
    }

    public StompFrame receive() throws IOException {
        return receive(RECEIVE_TIMEOUT);
    }

    public StompFrame receive(long timeOut) throws IOException {
        stompSocket.setSoTimeout((int) timeOut);
        InputStream is = stompSocket.getInputStream();
        StompWireFormat wf = new StompWireFormat();
        DataInputStream dis = new DataInputStream(is);
        return (StompFrame) wf.unmarshal(dis);
    }

    public String receiveFrame() throws IOException {
        return receiveFrame(RECEIVE_TIMEOUT);
    }

    public String receiveFrame(long timeOut) throws IOException {
        stompSocket.setSoTimeout((int) timeOut);
        InputStream is = stompSocket.getInputStream();
        int c = 0;
        for (;;) {
            c = is.read();
            if (c < 0) {
                throw new IOException("socket closed.");
            } else if (c == 0) {
                c = is.read();
                if (c == '\n') {
                    // end of frame
                    return stringFromBuffer(inputBuffer);
                } else {
                    inputBuffer.write(0);
                    inputBuffer.write(c);
                }
            } else {
                inputBuffer.write(c);
            }
        }
    }

    public boolean isConnected() {
        if (stompSocket == null || !stompSocket.isConnected()) {
            connected = false;
        }

        return connected;
    }

    private String stringFromBuffer(ByteArrayOutputStream inputBuffer)
            throws IOException {
        byte[] ba = inputBuffer.toByteArray();
        inputBuffer.reset();
        return new String(ba, "UTF-8");
    }

    public Socket getStompSocket() {
        return stompSocket;
    }

    public void setStompSocket(Socket stompSocket) {
        this.stompSocket = stompSocket;
    }

    public void connect(String username, String password) throws IOException,
            MessageBusException {
        connect(username, password, null);
    }

    public void connect(String username, String password, String client)
            throws IOException, MessageBusException {
        Map<String, String> headers = new HashMap();
        headers.put("login", username);
        headers.put("passcode", password);
        if (client != null) {
            headers.put("client-id", client);
        }
        StompFrame frame = new StompFrame("CONNECT", headers);
        sendFrame(frame.format());

        StompFrame connect = receive();
        if (!connect.getAction().equals(Stomp.Responses.CONNECTED)) {
            throw new MessageBusException("Not connected to server: "
                    + connect.getBody());
        }
    }

    public void disconnect() throws IOException {

        StompFrame frame = new StompFrame("DISCONNECT");
        sendFrame(frame.format());
    }

    public void sendSafe(String destination, String message,
            Map<String, String> headers) throws IOException,
            SendFailedException {
        sendSafe(destination, message.getBytes(), headers);
    }

    public void sendSafe(String destination, byte[] message,
            Map<String, String> headers) throws IOException,
            SendFailedException {
        // Add receipt header to the headers.
        if (headers == null) {
            headers = new HashMap<String, String>();
        }
        headers.put("receipt", "1");

        // send the message.
        send(destination, message, headers);

        // Wait for the
        StompFrame connect = receive();
        if (!connect.getAction().equals(Stomp.Responses.RECEIPT)) {
            throw new SendFailedException("Failed to receive RECEIPT: "
                    + connect.getBody());
        }
        log.info("Message sent receipt acknowledged for destination:"
                + destination);
    }

    public void send(String destination, String message,
            Map<String, String> headers) throws IOException {
        send(destination, message.getBytes(), null, headers);
    }

    public void send(String destination, byte[] message,
            Map<String, String> headers) throws IOException {
        send(destination, message, null, headers);
    }

    public void send(String destination, byte[] message, String transaction,
            Map<String, String> headers) throws IOException {
        if (headers == null) {
            headers = new HashMap<String, String>();
        }
        headers.put("destination", destination);
        headers.put("persistent", "true");
        if (transaction != null) {
            headers.put("transaction", transaction);
        }

        StompFrame frame = new StompFrame("SEND", headers, message);
        sendFrame(frame.format());
    }

    public void subscribe(String destination) throws IOException {
        subscribe(destination, null, null);
    }

    public void subscribe(String destination, String ack) throws IOException {
        subscribe(destination, ack, new HashMap<String, String>());
    }

    public void subscribe(String destination, String ack,
            Map<String, String> headers) throws IOException{
        if (headers == null) {
            headers = new HashMap<String, String>();
        }
        headers.put("destination", destination);

        if (ack != null) {
            headers.put("ack", ack);
        }
        StompFrame frame = new StompFrame("SUBSCRIBE", headers);
        sendFrame(frame.format());
    }

    public void unsubscribe(String destination) throws IOException {
        unsubscribe(destination, null);
    }

    public void unsubscribe(String destination, Map<String, String> headers)
            throws IOException {
        if (headers == null) {
            headers = new HashMap<String, String>();
        }
        headers.put("destination", destination);

        StompFrame frame = new StompFrame("UNSUBSCRIBE", headers);
        sendFrame(frame.format());
    }

    public void begin(String transaction) throws IOException {
        Map<String, String> headers = new HashMap<String, String>();
        headers.put("transaction", transaction);
        StompFrame frame = new StompFrame("BEGIN", headers);
        sendFrame(frame.format());
    }

    public void abort(String transaction) throws IOException {
        Map<String, String> headers = new HashMap<String, String>();
        headers.put("transaction", transaction);
        StompFrame frame = new StompFrame("ABORT", headers);
        sendFrame(frame.format());
    }

    public void commit(String transaction) throws IOException {
        Map<String, String> headers = new HashMap<String, String>();
        headers.put("transaction", transaction);
        StompFrame frame = new StompFrame("COMMIT", headers);
        sendFrame(frame.format());
    }

    public void ack(String messageId) throws IOException {
        ack(messageId, null, null, null, null);
    }

    public void ack(String messageId, String receiptId) throws IOException {
        ack(messageId, null, null, null, receiptId);
    }

    public void ack(String messageId, String transaction,
            String subscriptionId, String connectionId, String receiptId)
            throws IOException {
        Map<String, String> headers = new HashMap<String, String>();
        headers.put("message-id", messageId);
        log.debug("acking message-id: " + messageId);
        if (transaction != null)
            headers.put("transaction", transaction);
        if (subscriptionId != null && !subscriptionId.equals(""))
            headers.put("subscription", subscriptionId);

        if (connectionId != null && !connectionId.equals(Utils.NULL_STRING))
            headers.put("connection-id", connectionId);

        if (receiptId != null && !receiptId.equals(Utils.NULL_STRING))
            headers.put("receipt", receiptId);

        StompFrame frame = new StompFrame("ACK", headers);
        sendFrame(frame.format());
    }

    public void keepAlive() throws IOException {
        Map<String, String> headers = new HashMap<String, String>();
        StompFrame frame = new StompFrame("KEEPALIVE", headers);
        sendFrame(frame.format());
    }

    public void credit(StompFrame frame) throws IOException {
        credit(frame.getHeaders().get("message-id"));
    }

    public void credit(String messageId) throws IOException {
        Map<String, String> headers = new HashMap<String, String>();
        headers.put("message-id", messageId);
        StompFrame frame = new StompFrame("CREDIT", headers);
        sendFrame(frame.format());
    }

    protected String appendHeaders(HashMap<String, Object> headers) {
        StringBuffer result = new StringBuffer();
        for (String key : headers.keySet()) {
            result.append(key + ":" + headers.get(key) + "\n");
        }
        result.append("\n");
        return result.toString();
    }

    public void nack(String messageId, String subscriptionId)
            throws IOException {
        Map<String, String> headers = new HashMap<String, String>();
        headers.put("message-id", messageId);
        if (subscriptionId != null && !subscriptionId.equals(""))
            headers.put("subscription", subscriptionId);

        StompFrame frame = new StompFrame("NACK", headers);
        sendFrame(frame.format());
    }

}


text:
function SmartInventoryEdit($scope, $location,
    $stateParams, InventoryForm, Rest, ProcessErrors,
    GetBasePath, ParseTypeChange, Wait, ToJSON,
    ParseVariableString, $state, OrgAdminLookup, resourceData,
    $rootScope, InstanceGroupsService, InstanceGroupsData) {

    // Inject dynamic view
    var defaultUrl = GetBasePath('inventory'),
        form = InventoryForm,
        inventory_id = $stateParams.smartinventory_id,
        inventoryData = resourceData.data,
        instance_group_url = inventoryData.related.instance_groups;
    init();

    function init() {
        form.formLabelSize = null;
        form.formFieldSize = null;
        $scope.inventory_id = inventory_id;

        $scope = angular.extend($scope, inventoryData);

        $scope.smartinventory_variables = inventoryData.variables === null || inventoryData.variables === '' ? '---' : ParseVariableString(inventoryData.variables);
        $scope.organization_name = inventoryData.summary_fields.organization.name;
        $scope.instance_groups = InstanceGroupsData;

        $scope.$watch('inventory_obj.summary_fields.user_capabilities.edit', function(val) {
            if (val === false) {
                $scope.canAdd = false;
            }
        });

        $scope.parseType = 'yaml';

        $scope.inventory_obj = inventoryData;
        $rootScope.breadcrumb.inventory_name = inventoryData.name;

        ParseTypeChange({
            scope: $scope,
            variable: 'smartinventory_variables',
            parse_variable: 'parseType',
            field_id: 'smartinventory_smartinventory_variables',
            readOnly: !$scope.inventory_obj.summary_fields.user_capabilities.edit
        });

        OrgAdminLookup.checkForAdminAccess({organization: inventoryData.organization})
        .then(function(canEditOrg){
            $scope.canEditOrg = canEditOrg;
        });

        $scope.smart_hosts = {
            host_filter: encodeURIComponent($scope.host_filter)
        };
    }

    // Save
    $scope.formSave = function() {
        Wait('start');

        let fld, data = {};

        for (fld in form.fields) {
            data[fld] = $scope[fld];
        }

        data.variables = ToJSON($scope.parseType, $scope.smartinventory_variables, true);
        data.host_filter = decodeURIComponent($scope.smart_hosts.host_filter);
        data.kind = "smart";

        Rest.setUrl(defaultUrl + inventory_id + '/');
        Rest.put(data)
            .then(() => {
                InstanceGroupsService.editInstanceGroups(instance_group_url, $scope.instance_groups)
                    .then(() => {
                        Wait('stop');
                        $state.go($state.current, {}, { reload: true });
                    })
                    .catch(({data, status}) => {
                        ProcessErrors($scope, data, status, form, {
                            hdr: 'Error!',
                            msg: 'Failed to update instance groups. POST returned status: ' + status
                        });
                    });
            })
            .catch(({data, status}) => {
                ProcessErrors($scope, data, status, form, {
                    hdr: 'Error!',
                    msg: 'Failed to update inventory. PUT returned status: ' + status
                });
            });
    };

    $scope.formCancel = function() {
        $state.go('inventories');
    };

}

export default [ '$scope', '$location',
    '$stateParams', 'InventoryForm', 'Rest',
    'ProcessErrors', 'GetBasePath', 'ParseTypeChange', 'Wait',
    'ToJSON', 'ParseVariableString',
    '$state', 'OrgAdminLookup', 'resourceData',
    '$rootScope', 'InstanceGroupsService', 'InstanceGroupsData', SmartInventoryEdit
];

redpajama_set_name: RedPajamaGithub


text:
package cloudwatch

import (
	"errors"
	"html/template"
	"reflect"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/service/cloudwatch"
	"github.com/aws/aws-sdk-go/service/cloudwatch/cloudwatchiface"

	"istio.io/istio/mixer/adapter/cloudwatch/config"
	"istio.io/istio/mixer/pkg/adapter"
	"istio.io/istio/mixer/pkg/adapter/test"
	"istio.io/istio/mixer/template/metric"
)

type mockCloudWatchClient struct {
	cloudwatchiface.CloudWatchAPI
}

func (m *mockCloudWatchClient) PutMetricData(input *cloudwatch.PutMetricDataInput) (*cloudwatch.PutMetricDataOutput, error) {
	return &cloudwatch.PutMetricDataOutput{}, nil
}

type mockFailCloudWatchClient struct {
	cloudwatchiface.CloudWatchAPI
}

func (m *mockFailCloudWatchClient) PutMetricData(input *cloudwatch.PutMetricDataInput) (*cloudwatch.PutMetricDataOutput, error) {
	return nil, errors.New("put metric data failed")
}

func TestPutMetricData(t *testing.T) {
	cfg := &config.Params{
		Namespace: "istio-mixer-cloudwatch",
	}

	env := test.NewEnv(t)

	logEntryTemplates := make(map[string]*template.Template)
	logEntryTemplates["inst"], _ = template.New("inst").Parse(`{{or (.sourceIp) "-"}} - {{or (.sourceUser) "-"}}`)

	cases := []struct {
		metricData        []*cloudwatch.MetricDatum
		expectedErrString string
		handler           adapter.Handler
	}{
		{
			[]*cloudwatch.MetricDatum{
				{MetricName: aws.String("testMetric"), Value: aws.Float64(1)},
			},
			"put metric data",
			newHandler(nil, nil, logEntryTemplates, env, cfg, &mockFailCloudWatchClient{}, &mockLogsClient{}),
		},
		{
			[]*cloudwatch.MetricDatum{
				{MetricName: aws.String("testMetric"), Value: aws.Float64(1)},
			},
			"",
			newHandler(nil, nil, logEntryTemplates, env, cfg, &mockCloudWatchClient{}, &mockLogsClient{}),
		},
	}

	for _, c := range cases {
		h, ok := c.handler.(*handler)
		if !ok {
			t.Error("Test case has the wrong type of handler.")
		}

		err := h.putMetricData(c.metricData)
		if len(c.expectedErrString) > 0 {
			if err == nil {
				t.Errorf("Expected an error containing %s, got none", c.expectedErrString)

			} else if !strings.Contains(strings.ToLower(err.Error()), c.expectedErrString) {
				t.Errorf("Expected an error containing \"%s\", actual error: \"%v\"", c.expectedErrString, err)
			}
		} else if err != nil {
			t.Errorf("Unexpected error: \"%v\"", err)
		}
	}
}

func generateCfgWithUnit(unit config.Params_MetricDatum_Unit) *config.Params {
	return generateCfgWithNameAndUnit("requestDuration", unit)
}

func generateCfgWithNameAndUnit(metricName string, unit config.Params_MetricDatum_Unit) *config.Params {
	return &config.Params{
		Namespace: "istio-mixer-cloudwatch",
		MetricInfo: map[string]*config.Params_MetricDatum{
			metricName: {
				Unit: unit,
			},
		},
	}
}

func TestDurationNumericValue(t *testing.T) {
	cases := []struct {
		config        *config.Params
		inst          *metric.Instance
		expectedValue float64
	}{
		{
			generateCfgWithUnit(config.None),
			&metric.Instance{
				Name:  "requestDuration",
				Value: 1 * time.Minute},
			60000,
		},
		{
			generateCfgWithUnit(config.Seconds),
			&metric.Instance{
				Name:  "requestDuration",
				Value: 1 * time.Minute},
			60,
		},
		{
			generateCfgWithUnit(config.Milliseconds),
			&metric.Instance{
				Name:  "requestDuration",
				Value: 1 * time.Minute},
			60000,
		},
		{
			generateCfgWithUnit(config.Microseconds),
			&metric.Instance{
				Name:  "requestDuration",
				Value: 1 * time.Minute},
			60000000,
		},
	}

	for _, c := range cases {
		v, ok := c.inst.Value.(time.Duration)
		if !ok {
			t.Error("Test case instance value is of the wrong type")
		}

		unit := c.config.GetMetricInfo()["requestDuration"].GetUnit()
		n := getDurationNumericValue(v, unit)

		if n != c.expectedValue {
			t.Errorf("Expected value %v, got %v", c.expectedValue, n)
		}

	}
}

func TestGetNumericValue(t *testing.T) {
	cases := []struct {
		inst              *metric.Instance
		config            *config.Params
		expectedErrString string
		expectedValue     float64
	}{
		{&metric.Instance{Name: "invalid", Value: "invalidValue"},
			generateCfgWithNameAndUnit("invalid", config.None),
			"can't parse string", 0},
		{&metric.Instance{Name: "requestcount", Value: "1"},
			generateCfgWithNameAndUnit("requestcount", config.Count),
			"", 1.0},
		{&metric.Instance{Name: "requestcount", Value: 1},
			generateCfgWithNameAndUnit("requestcount", config.Count),
			"", 1.0},
		{&metric.Instance{Name: "requestcount", Value: 1.0},
			generateCfgWithNameAndUnit("requestcount", config.Count),
			"", 1.0},
		{&metric.Instance{Name: "requestcount", Value: true},
			generateCfgWithNameAndUnit("requestcount", config.Count),
			"unsupported value type", 0},
		{&metric.Instance{Name: "requestduration", Value: 1 * time.Minute},
			generateCfgWithNameAndUnit("requestduration", config.Seconds), "", 60},
	}

	for _, c := range cases {
		v, err := getNumericValue(c.inst.Value, c.config.GetMetricInfo()[c.inst.Name].GetUnit())
		if len(c.expectedErrString) == 0 && err != nil {
			t.Errorf("Unexpected error: \"%v\"", err)
		}

		if len(c.expectedErrString) > 0 {
			if err == nil {
				t.Errorf("Expected an error containing \"%s\", but no error was thrown", c.expectedErrString)

			} else if !strings.Contains(strings.ToLower(err.Error()), c.expectedErrString) {
				t.Errorf("Expected an error containing \"%s\", actual error: \"%v\"", c.expectedErrString, err)
			}

		} else {
			if v != c.expectedValue {
				t.Errorf("Expected value %v, got %v", c.expectedValue, v)
			}
		}
	}
}

func TestSendMetricsToCloudWatch(t *testing.T) {
	env := test.NewEnv(t)
	cfg := &config.Params{
		Namespace: "istio-mixer-cloudwatch",
	}
	logEntryTemplates := make(map[string]*template.Template)
	logEntryTemplates["inst"], _ = template.New("inst").Parse(`{{or (.sourceIp) "-"}} - {{or (.sourceUser) "-"}}`)

	h := newHandler(nil, nil, logEntryTemplates, env, cfg, &mockCloudWatchClient{}, &mockLogsClient{})

	cases := []struct {
		metricData                  []*cloudwatch.MetricDatum
		expectedCloudWatchCallCount int
	}{
		{[]*cloudwatch.MetricDatum{}, 0},
		{generateTestMetricData(1), 1},
		{generateTestMetricData(21), 2},
	}

	for _, c := range cases {
		hdr, ok := h.(*handler)
		if !ok {
			t.Error("Test case has the wrong type of handler.")
		}

		count, err := hdr.sendMetricsToCloudWatch(c.metricData)
		if err != nil {
			t.Errorf("Unexpected error: \"%v\"", err)
		}
		if count != c.expectedCloudWatchCallCount {
			t.Errorf("Expected %v calls but got %v", c.expectedCloudWatchCallCount, count)
		}
	}
}

func generateTestMetricData(count int) []*cloudwatch.MetricDatum {
	metricData := make([]*cloudwatch.MetricDatum, 0, count)

	for i := 0; i < count; i++ {
		metricData = append(metricData, &cloudwatch.MetricDatum{
			MetricName: aws.String("testMetric" + strconv.Itoa(i)),
			Value:      aws.Float64(1),
		})
	}

	return metricData
}

func TestGenerateMetricData(t *testing.T) {
	env := test.NewEnv(t)
	logEntryTemplates := make(map[string]*template.Template)
	logEntryTemplates["inst"], _ = template.New("inst").Parse(`{{or (.sourceIp) "-"}} - {{or (.sourceUser) "-"}}`)
	cases := []struct {
		handler            adapter.Handler
		insts              []*metric.Instance
		expectedMetricData []*cloudwatch.MetricDatum
	}{
		// empty instances
		{
			newHandler(nil, nil, logEntryTemplates, env,
				generateCfgWithUnit(config.Count),
				&mockCloudWatchClient{}, &mockLogsClient{}),
			[]*metric.Instance{}, []*cloudwatch.MetricDatum{}},
		// timestamp value
		{
			newHandler(nil, nil, logEntryTemplates, env,
				generateCfgWithNameAndUnit("requestduration", config.Milliseconds),
				&mockCloudWatchClient{}, &mockLogsClient{}),
			[]*metric.Instance{
				{
					Value: 1 * time.Minute,
					Name:  "requestduration",
				},
			},
			[]*cloudwatch.MetricDatum{
				{
					MetricName: aws.String("requestduration"),
					Unit:       aws.String(config.Milliseconds.String()),
					Value:      aws.Float64(60000),
					Dimensions: []*cloudwatch.Dimension{},
				},
			},
		},
		// count value and dimensions
		{
			newHandler(nil, nil, logEntryTemplates, env,
				generateCfgWithNameAndUnit("requestcount", config.Count),
				&mockCloudWatchClient{}, &mockLogsClient{}),
			[]*metric.Instance{
				{
					Value: "1",
					Name:  "requestcount",
					Dimensions: map[string]interface{}{
						"arbitraryDimension": 50,
					},
				},
			},
			[]*cloudwatch.MetricDatum{
				{
					MetricName: aws.String("requestcount"),
					Unit:       aws.String(config.Count.String()),
					Value:      aws.Float64(1),
					Dimensions: []*cloudwatch.Dimension{
						{
							Name:  aws.String("arbitraryDimension"),
							Value: aws.String("50"),
						},
					},
				},
			},
		},
	}

	for _, c := range cases {
		h, ok := c.handler.(*handler)
		if !ok {
			t.Error("Test case has the wrong type of handler.")
		}

		md := h.generateMetricData(c.insts)

		if len(c.expectedMetricData) != len(md) {
			t.Errorf("Expected %v metric data items but got %v", len(c.expectedMetricData), len(md))
		}

		for i := 0; i < len(c.expectedMetricData); i++ {
			expectedMD := c.expectedMetricData[i]
			actualMD := md[i]

			if !reflect.DeepEqual(expectedMD, actualMD) {
				t.Errorf("Expected %v, actual %v", expectedMD, actualMD)
			}
		}
	}
}


text:
define(["require", "exports", "esri/widgets/support/widget"], function (require, exports, widget_1) {
    "use strict";
    Object.defineProperty(exports, "__esModule", { value: true });
    exports.default = function (props) {
        var author = props.config.showAuthor ? (widget_1.jsxFactory.createElement("p", { class: "font-size--1 card-last hug-bottom", key: props.item.title + "-author" }, props.item.owner)) : null;
        var PanelBaseComponent = {
            captionOpacity: 1,
            captionTransform: 0,
            render: function () {
                return (widget_1.jsxFactory.createElement("div", { class: "card block trailer-1 animate-fade-in card-fade", style: "background-color: " + props.config.cardColor + ";", key: props.item.title + "-div" },
                    widget_1.jsxFactory.createElement("figure", { class: "card-image-wrap" },
                        widget_1.jsxFactory.createElement("a", { title: props.i18n.ui.galleryTip, role: "link", tabindex: "0", onkeypress: handleKeyPress },
                            widget_1.jsxFactory.createElement("img", { class: "card-image clickable thumbnail-min", img: true, src: "data:image/gif;base64,R0lGODlhAQABAIAAAAAAAP///yH5BAEAAAAALAAAAAABAAEAAAIBRAA7", alt: props.item.title, onmouseover: handleMouseOver, onmouseleave: handleMouseOut, onclick: props.itemClickHandler, style: "\n                                    background-image: url(" + props.item.thumbnailUrl + ");\n                                    background-repeat: no-repeat;\n                                    background-size: cover;\n                                " })),
                        widget_1.jsxFactory.createElement("div", { class: "card-image-caption", style: "\n                                opacity: " + PanelBaseComponent.captionOpacity + ";\n                                transform: translate(0, " + PanelBaseComponent.captionTransform + "%);\n                                background-color: " + convertHex(props.captionColor, 80) + ";\n                                color: " + props.config.captionTextColor + "\n                            " }, props.item.displayName)),
                    widget_1.jsxFactory.createElement("div", { class: "card-content", style: "color: " + props.config.fontColor },
                        widget_1.jsxFactory.createElement("a", { title: props.i18n.ui.galleryTip, style: "color: " + props.config.linkColor },
                            widget_1.jsxFactory.createElement("h5", { tabindex: "0", class: "clickable", onclick: props.itemClickHandler }, props.item.title)),
                        author,
                        widget_1.jsxFactory.createElement("div", { class: "open-out-container" },
                            widget_1.jsxFactory.createElement("a", { class: "open-out-icon btn btn-transparent icon-ui-description", title: props.extItem, href: props.extLink, style: "color: " + props.config.buttonBgColor, key: props.item.title + "-info-icon" }),
                            widget_1.jsxFactory.createElement("a", { class: "open-out-icon btn btn-transparent icon-ui-maximize", title: props.extTitle, href: props.maxLink, style: "color: " + props.config.buttonBgColor, key: props.item.title + "-open-out-icon" })))));
            }
        };
        function handleMouseOver() {
            PanelBaseComponent.captionOpacity = 0;
            PanelBaseComponent.captionTransform = 100;
        }
        function handleMouseOut() {
            PanelBaseComponent.captionOpacity = 1;
            PanelBaseComponent.captionTransform = 0;
        }
        function handleKeyPress(e) {
            if (e.keyCode === 13) {
                props.itemClickHandler();
            }
        }
        function convertHex(hex, opacity) {
            hex = hex.replace("#", "");
            var r = parseInt(hex.substring(0, 2), 16);
            var g = parseInt(hex.substring(2, 4), 16);
            var b = parseInt(hex.substring(4, 6), 16);
            var result = "rgba(" + r + "," + g + "," + b + "," + opacity / 100 + ")";
            return result;
        }
        return PanelBaseComponent;
    };
});
//# sourceMappingURL=PanelBase.js.map


text:
using System;
using System.Collections.Generic;
using System.Diagnostics.CodeAnalysis;
using DigiRazor.PasswordValidation.Factories;
using DigiRazor.PasswordValidation.Model;
using FluentAssertions;
using Microsoft.VisualStudio.TestTools.UnitTesting;

namespace DigiRazor.PasswordValidation.UnitTests
{
    [TestClass]
    [ExcludeFromCodeCoverage]
    public class PasswordServiceTest
    {
        private IPassword validTestPassword;

        private IValidatorFactory factory;
        private PasswordRules validTestRules;

        private IPasswordService service;

        [TestInitialize]
        public void Initialise()
        {
            validTestPassword = new Password
            {
                UserId = "ABHW089",
                OldPassword = "B1ge@rs*",
                NewPassword = "yVHn6?R@",
                ConfirmPassword = "yVHn6?R@",
                NewPasswordHash = "yVHn6?R@",

                IsValid = true
            };

            validTestPassword.SetHistory(new List<string> { "$sG96r#X", "3g9m&9W7" });

            validTestRules = new PasswordRules
            {
                Validators = ValidatorTypes.All,
                MinLength = 8,
                MaxLength = 10,
                SpecialChars = new[] { '!', '@', '#', '$', '%', '*', '+', '/' },
                MinHistory = 3,
                Blacklist = new[] { "test", "password" }
            };

            factory = new ValidatorFactory();

            service = new PasswordService(factory);
        }

        [TestCategory("Unit")]
        [TestCategory("Unit-PasswordService")]
        [TestMethod]
        public void Test_PasswordService_No_Injected_Factory()
        {
            var serv = new PasswordService();
            serv.SetupRules(validTestRules);

            var ruleSet = serv.RuleSet;
            ruleSet.Should().NotBeNull();

            var result = serv.Validate(validTestPassword);

            result.Should().NotBe(null);
            result.IsValid.Should().BeTrue();
        }

        [TestCategory("Unit")]
        [TestCategory("Unit-PasswordService")]
        [TestMethod]
        public void Test_PasswordService_SetupRules()
        {
            service.SetupRules(validTestRules);

            var result = service.RuleSet;

            result.Should().NotBeNull();
        }

        [TestCategory("Unit")]
        [TestCategory("Unit-PasswordService")]
        [TestMethod]
        public void Test_PasswordService_SetupRules_Null_Parameter()
        {

            var result = new Action(() =>
            {
                service.SetupRules(null);

            });

            result.ShouldThrow<ArgumentNullException>();
        }

        [TestCategory("Unit")]
        [TestCategory("Unit-PasswordService")]
        [TestMethod]
        public void Test_PasswordService_Validate()
        {
            service.SetupRules(validTestRules);

            var result = service.Validate(validTestPassword);

            result.Should().NotBe(null);
            result.IsValid.Should().BeTrue();
        }

        [TestCategory("Unit")]
        [TestCategory("Unit-PasswordService")]
        [TestMethod]
        public void Test_PasswordService_Validate_Null_Parameter()
        {
            service.SetupRules(validTestRules);

            var result = new Action(() =>
            {
                service.Validate(null);

            });

            result.ShouldThrow<ArgumentNullException>();
        }

        [TestCategory("Unit")]
        [TestCategory("Unit-PasswordService")]
        [TestMethod]
        public void Test_PasswordService_Fail_On_ConfirmPassword()
        {
            var testPassword = new Password
            {
                UserId = "ABHW089",
                OldPassword = "B1ge@rs*",
                NewPassword = "yVHn6?R@",
                ConfirmPassword = "yVHn6?R2",
                NewPasswordHash = "yVHn6?R@",

                IsValid = true
            };

            testPassword.SetHistory(new List<string> { "$sG96r#X", "3g9m&9W7" });

            service.SetupRules(validTestRules);

            var result = service.Validate(testPassword);

            result.Should().NotBe(null);
            result.IsValid.Should().BeFalse();
        }

        [TestCategory("Unit")]
        [TestCategory("Unit-PasswordService")]
        [TestMethod]
        public void Test_PasswordService_AddCustomValidator()
        {
            service.SetupRules(validTestRules);

            var result = new Action(() =>
            {
                service.AddCustomValidator(new CustomValidator1());

            });

            result.ShouldNotThrow<ArgumentException>();
        }

        [TestCategory("Unit")]
        [TestCategory("Unit-PasswordService")]
        [TestMethod]
        public void Test_PasswordService_AddCustomValidator_Null_Parameter()
        {
            service.SetupRules(validTestRules);

            var result = new Action(() =>
            {
                service.AddCustomValidator(null);

            });

            result.ShouldThrow<ArgumentNullException>();
        }

        [TestCategory("Unit")]
        [TestCategory("Unit-PasswordService")]
        [TestMethod]
        public void Test_PasswordService_AddCustomValidator_InValid()
        {
            service.SetupRules(validTestRules);

            var result = new Action(() =>
            {
                service.AddCustomValidator(new CustomValidator2());

            });

            result.ShouldThrow<ArgumentException>();
        }

    }

    [ExcludeFromCodeCoverage]
    public class CustomValidator1 : IPasswordValidator
    {

        public ValidatorTypes Type
        {
            get { return ValidatorTypes.Custom; }
        }

        public void Setup(PasswordRules ruleSet)
        {

        }

        public IPassword Validate(IPassword value)
        {
            value.IsValid = true;

            return value;
        }

        public string ToString(PasswordRules ruleSet)
        {
            return "Custom Validator 1";
        }
    }

    [ExcludeFromCodeCoverage]
    public class CustomValidator2 : IPasswordValidator
    {

        public ValidatorTypes Type
        {
            get { return ValidatorTypes.Blacklist; }
        }

        public void Setup(PasswordRules ruleSet)
        {

        }

        public IPassword Validate(IPassword value)
        {
            value.IsValid = true;

            return value;
        }

        public string ToString(PasswordRules ruleSet)
        {
            return "Custom Validator 2";
        }
    }
}


text:
<?php

defined('_JEXEC') or die;

JLoader::register('NewsfeedsHelper', JPATH_ADMINISTRATOR . '/components/com_newsfeeds/helpers/newsfeeds.php');

/**
 * Utility class for creating HTML Grids.
 *
 * @since  1.5
 */
class JHtmlNewsfeed
{
	/**
	 * Get the associated language flags
	 *
	 * @param   int  $newsfeedid  The item id to search associations
	 *
	 * @return  string  The language HTML
	 *
	 * @throws  Exception  Throws a 500 Exception on Database failure
	 */
	public static function association($newsfeedid)
	{
		// Defaults
		$html = '';

		// Get the associations
		if ($associations = JLanguageAssociations::getAssociations('com_newsfeeds', '#__newsfeeds', 'com_newsfeeds.item', $newsfeedid))
		{
			foreach ($associations as $tag => $associated)
			{
				$associations[$tag] = (int) $associated->id;
			}

			// Get the associated newsfeed items
			$db = JFactory::getDbo();
			$query = $db->getQuery(true)
				->select('c.id, c.name as title')
				->select('l.sef as lang_sef')
				->from('#__newsfeeds as c')
				->select('cat.title as category_title')
				->join('LEFT', '#__categories as cat ON cat.id=c.catid')
				->where('c.id IN (' . implode(',', array_values($associations)) . ')')
				->join('LEFT', '#__languages as l ON c.language=l.lang_code')
				->select('l.image')
				->select('l.title as language_title');
			$db->setQuery($query);

			try
			{
				$items = $db->loadObjectList('id');
			}
			catch (RuntimeException $e)
			{
				throw new Exception($e->getMessage(), 500);
			}

			if ($items)
			{
				foreach ($items as &$item)
				{
					$text = strtoupper($item->lang_sef);
					$url = JRoute::_('index.php?option=com_newsfeeds&task=newsfeed.edit&id=' . (int) $item->id);
					$tooltipParts = array(
						JHtml::_('image', 'mod_languages/' . $item->image . '.gif',
							$item->language_title,
							array('title' => $item->language_title),
							true
						),
						$item->title,
						'(' . $item->category_title . ')'
					);
					$item->link = JHtml::_(
						'tooltip',
						implode(' ', $tooltipParts),
						null,
						null,
						$text,
						$url,
						null,
						'hasTooltip label label-association label-' . $item->lang_sef
					);
				}
			}

			$html = JLayoutHelper::render('joomla.content.associations', $items);
		}

		return $html;
	}
}


text:
{-# LANGUAGE NoNumericUnderscores #-}

-- Test for NumericUnderscores extension.
-- See #14473
-- This is a testcase for integer literal
-- in NO NumericUnderscores extension.

module NoNumericUnderscores0 where

f :: Int -> ()
f 1_000 = ()
f _   = ()


text:
package com.olegchir.flussonic_userlinks.wicket.IncludeResolver;

import java.util.*;

import org.apache.commons.lang3.StringUtils;
import org.apache.wicket.Application;
import org.apache.wicket.Component;
import org.apache.wicket.MarkupContainer;
import org.apache.wicket.WicketRuntimeException;
import org.apache.wicket.core.util.lang.PropertyResolver;
import org.apache.wicket.markup.ComponentTag;
import org.apache.wicket.markup.MarkupElement;
import org.apache.wicket.markup.MarkupException;
import org.apache.wicket.markup.MarkupStream;
import org.apache.wicket.markup.WicketTag;
import org.apache.wicket.markup.html.include.Include;
import org.apache.wicket.markup.parser.filter.WicketTagIdentifier;
import org.apache.wicket.markup.resolver.IComponentResolver;
import org.apache.wicket.model.Model;
import org.apache.wicket.request.Response;
import org.apache.wicket.response.StringResponse;
import org.apache.wicket.util.string.Strings;
import org.apache.wicket.util.string.interpolator.MapVariableInterpolator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Resolver of wicket:include tag
 *
 * Originally from: http://sanityresort.blogspot.ru/2011/08/creating-custom-wicket-tag-resolver.html
 * Adopted by: Oleg Chirukhin
 *
 * This class is a tag resolver to handle tags of the following format
 * <wicket:include [key="xxx"|filename="xxx"]/>.
 *
 *
 * This class tries to provide the same functionality for included html files as
 * WicketMessageResolver does for property file entries but also tries to avoid
 * to fail early.
 *
 * You have to specify either a key or a filename. A key will be looked up via
 * the resource mechanism and the result will be used as filename.
 * In addition you can nest tags within this tag. Those child tags will then be
 * used to generate the replacements for variables defined in the file content.
 * If this fails for a variable the parent's model object will queried for a
 * property with the variable's value as name. Should this also fail the last
 * fall back is the same query on the parent object.
 *
 * If the resource settings are set to throw exceptions on missing properties
 * this class will raise an exception if either a variable cannot be looked up
 * or a child component is not replaced.
 *
 * This tag allows you to easily include files in your pages which can also
 * depend on your localized properties.
 *
 * Usage:
 *     <wicket:include key="myFile">
 *        This text will be replaced with text from the specified file.
 *        <span wicket:id="replaceme">[to be replaced]</span>.
 *     </wicket:include>
 *
 *
 * Then your property file needs an entry like this:
 *
 *
 * myFile = path/to/file
 *
 *
 * This file contains html fragments with variables:
 *
 *
 * This is a file with ${replaceme}.
 *
 *
 * And your java component add:
 *
 *
 * add(new Label("replaceme", new Model<String>("some replaced text")));
 *
 *
 * The output will be:
 *
 *
 * This is a file with some replaced text.
 *
 *
 * @author Chris
 *
 */
public class IncludeResolver implements IComponentResolver {

    /**
     *
     */
    private static final long serialVersionUID = -9164415653709941809L;

    private static final String tagname = "include";

    private static final Logger log = LoggerFactory.getLogger(IncludeResolver.class);

    // register the tagname
    static {
        WicketTagIdentifier.registerWellKnownTagName(tagname);
    }

    /**
     * Handles resolving the wicket:include tag. If a filename is specified this
     * file will be looked up. If a key is specified it's value is looked up
     * using the resource mechanism. The looked up value will then be used as
     * filename.
     *
     */
    public Component resolve(MarkupContainer container, MarkupStream markupStream, ComponentTag tag) {

        if ((tag instanceof WicketTag) && tagname.equalsIgnoreCase(tag.getName())) {
            WicketTag wtag = (WicketTag) tag;

            String fileName = StringUtils.trimToNull(wtag.getAttribute("file"));
            String fileKey = StringUtils.trimToNull(wtag.getAttribute("key"));

            if (null != fileKey) {
                if (null != fileName) {
                    throw new MarkupException( "Wrong format of: you must not use file and key attribtue at once");
                }
                fileName = StringUtils.trimToNull(container.getString(fileKey));
                if (null == fileName) {
                    throw new MarkupException("The key inside could not be resolved");
                }

            } else if (null == fileName) {
                throw new MarkupException(
                        "Wrong format of: specify the file or key attribute");
            }

            final String id = "_" + tagname + "_" + container.getPage().getAutoIndex();
            IncludeContainer ic = new IncludeContainer(id, fileName, fileKey, markupStream);
            ic.setRenderBodyOnly(container.getApplication().getMarkupSettings().getStripWicketTags());
//            container.autoAdd(ic, markupStream);

            return ic;

        }
        return null;
    }

    /**
     * Helper class to break open the original Include class by using
     * inheritance.
     *
     * @author Chris
     *
     */
    private static final class ExposingInclude extends Include {

        /**
         *
         */
        private static final long serialVersionUID = -212861726226482365L;

        public ExposingInclude(String id, String filename) {
            super(id, filename);
        }

        public final String expose() {
            return this.importAsString();
        }

    }

    /**
     * Container class to render the included file.
     *
     * @author Chris
     *
     */
    private static final class IncludeContainer extends MarkupContainer {

        /**
         *
         */
        private static final long serialVersionUID = 3991477945186422264L;
        private final String key;
        private final String fileName;

        public IncludeContainer(String id, String fileName, String key, MarkupStream markupStream) {
            super(id, new Model(fileName));
            this.key = key;
            this.fileName = fileName;

        }

        /**
         * Wrapper to log this instance's settings.
         *
         * @return String containing the instance's settings.
         */
        private final String logParams() {
            return " tag with " + (null != this.key ? "key = '" + this.key + "' and resulting " : "") + "fileName = '" + this.fileName + "'";
        }

        /**
         * Renders the component tag. The base content is read from the file
         * defined by the filename. Afterwards all containing variables are
         * replaced by the following methods in this order:
         *

         *
         trying to get the value from a child component that has the
         * variable's value as id
         *
         trying to get the value as property from the parent's model
         * object
         *
         trying to get the value as property from the parent object
         *

         *
         * If all attempts to lookup up variable fail or not all child tags are
         * replaced either an exception is thrown or a warning is logged
         * depending on the resource settings.
         *
         */
        @Override
        public void onComponentTagBody(final MarkupStream markupStream, final ComponentTag openTag) {

            // get all child tags' rendered contents
            Map children = this.getRenderedChildren(markupStream, openTag);

            // read the included file content
            String fileContent = new ExposingInclude(this.getId() + "_internalInclude", this.getDefaultModelObjectAsString()).expose();

            // remember all replaced child tags
            final Set replacedChildren = new HashSet();

            // substitute all variables and set the result as response
            this.getResponse().write(new MapVariableInterpolator(fileContent, children) {

                @Override
                protected String getValue(String variableName) {

                    // try to get value from rendered child tags
                    String value = super.getValue(variableName);

                    if (null != value) {
                        // if we used a child tag create a marker
                        replacedChildren.add(variableName);
                    } else {
                        if (log.isDebugEnabled()) {
                            log.debug(IncludeContainer.this.logParams()
                                    + " - Could not find replacement value for variable = '" + variableName
                                    + "' in child tags");
                        }
                        // see if the variable can be replaced using the
                        // parent's model object - use try/catch to avoid fail
                        // early
                        try {
                            value = Strings.toString(PropertyResolver.getValue(variableName, IncludeContainer.this.getParent().getDefaultModelObject()));
                        } catch (WicketRuntimeException e) {
                            if (log.isDebugEnabled()) {
                                log.debug(IncludeContainer.this.logParams()
                                        + " - Could not find replacement value for variable = '" + variableName
                                        + "' in parent model");
                            }
                        }
                    }

                    if (null == value) {
                        // see if the variable can be replaced using the parent
                        // object - use try/catch to avoid fail early
                        try {
                            value = Strings.toString(PropertyResolver.getValue(variableName, IncludeContainer.this.getParent()));
                        } catch (WicketRuntimeException e) {
                            if (log.isDebugEnabled()) {
                                log.debug(IncludeContainer.this.logParams()
                                        + " - Could not find replacement value for variable = '" + variableName
                                        + "' in parent object");
                            }
                        }
                    }

                    if (null == value) {
                        // Handle failed lookup
                        String logMsg = IncludeContainer.this.logParams()
                                + " - Bailing with exception since variable = '" + variableName
                                + "' could not be replaced";

                        if (IncludeContainer.this.isThrowingExceptions()) {
                            if (log.isDebugEnabled()) {
                                log.debug(logMsg);

                            }
                            markupStream.throwMarkupException(IncludeContainer.this.logParams()
                                    + " - Could not replace variable  = '" + variableName + "'");

                        } else {
                            log.warn(logMsg);
                        }
                    }

                    return value;
                }

            }.toString());

            // Make sure all of the children were rendered
            Iterator<String> iter = children.keySet().iterator();
            while (iter.hasNext()) {
                String id = iter.next();
                if (replacedChildren.contains(id) == false) {
                    String msg = "The  for file " + this.fileName
                            + "has a child element with wicket:id=\"" + id + "\". You must add the variable ${" + id
                            + "} to the file content for the wicket:include.";

                    if (this.isThrowingExceptions() == true) {
                        markupStream.throwMarkupException(msg);
                    } else {
                        log.warn(msg);
                    }
                }
            }
        }

        /**
         * Wraps call to resource settings to determine if a exception should be
         * thrown in case not all variables are resolved or not all children are
         * rendered.
         *
         * @return
         *         Application.get().getResourceSettings().getThrowExceptionOnMissingResource()
         */
        private final boolean isThrowingExceptions() {
            return Application.get().getResourceSettings().getThrowExceptionOnMissingResource();
        }

        /**
         * Walks through all children of the current tag and stores their
         * rendering results in a map.
         *
         * @param markupStream
         *            The markupStream associated with the tag.
         * @param openTag
         *            The current include tag to be rendered.
         * @return Map containing all children's rendered responses associated
         *         with each child's id.
         */
        private Map<String, String> getRenderedChildren(final MarkupStream markupStream, final ComponentTag openTag) {
            Map<String, String> children = new HashMap<>();

            if (!openTag.isOpenClose()) {

                while (markupStream.hasMore() && !markupStream.get().closes(openTag)) {
                    MarkupElement element = markupStream.get();
                    // If it a tag like  or
                    if ((element instanceof ComponentTag) && !markupStream.atCloseTag()) {
                        String id = ((ComponentTag) element).getId();

                        Component comp = this.getParent().get(id);
                        if (comp != null) {
                            children.put(id, this.getRenderedResponseString(markupStream, comp));
//                            markupStream.next();
                        } else {
                            markupStream.next();
                        }
                    } else {
                        markupStream.next();
                    }
                }

            }

            return children;
        }

        /**
         * Obtains the rendered response of the given component rendered
         * according to the markupStream. Rendering uses a temporary response
         * and the original response is restored before returning the result.
         *
         * @param markupStream
         *            The markupStream used to render the component.
         * @param comp
         *            The component to be rendered.
         * @return The rendered result.
         */
        private String getRenderedResponseString(final MarkupStream markupStream, Component comp) {
            Response webResponse = comp.getResponse();

            StringResponse response = new StringResponse();

            try {
                this.getRequestCycle().setResponse(response);
                comp.render();
                markupStream.skipComponent();
            } finally {
                this.getRequestCycle().setResponse(webResponse);
            }
            return response.getBuffer().toString();
        }
    }
}


text:
// Package vtgate provides query routing rpc services
// for vttablets.
package vtgate

import (
	"flag"
	"fmt"
	"math"
	"net/http"
	"time"

	log "github.com/golang/glog"
	"golang.org/x/net/context"

	"vitess.io/vitess/go/acl"
	"vitess.io/vitess/go/flagutil"
	"vitess.io/vitess/go/sqltypes"
	"vitess.io/vitess/go/stats"
	"vitess.io/vitess/go/tb"
	"vitess.io/vitess/go/vt/discovery"
	"vitess.io/vitess/go/vt/key"
	"vitess.io/vitess/go/vt/logutil"
	"vitess.io/vitess/go/vt/servenv"
	"vitess.io/vitess/go/vt/sqlannotation"
	"vitess.io/vitess/go/vt/sqlparser"
	"vitess.io/vitess/go/vt/srvtopo"
	"vitess.io/vitess/go/vt/topo"
	"vitess.io/vitess/go/vt/topo/topoproto"
	"vitess.io/vitess/go/vt/vterrors"

	"vitess.io/vitess/go/vt/vtgate/gateway"
	"vitess.io/vitess/go/vt/vtgate/vtgateservice"

	querypb "vitess.io/vitess/go/vt/proto/query"
	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
	vtgatepb "vitess.io/vitess/go/vt/proto/vtgate"
	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
)

var (
	transactionMode     = flag.String("transaction_mode", "MULTI", "SINGLE: disallow multi-db transactions, MULTI: allow multi-db transactions with best effort commit, TWOPC: allow multi-db transactions with 2pc commit")
	normalizeQueries    = flag.Bool("normalize_queries", true, "Rewrite queries with bind vars. Turn this off if the app itself sends normalized queries with bind vars.")
	streamBufferSize    = flag.Int("stream_buffer_size", 32*1024, "the number of bytes sent from vtgate for each stream call. It's recommended to keep this value in sync with vttablet's query-server-config-stream-buffer-size.")
	queryPlanCacheSize  = flag.Int64("gate_query_cache_size", 10000, "gate server query cache size, maximum number of queries to be cached. vtgate analyzes every incoming query and generate a query plan, these plans are being cached in a lru cache. This config controls the capacity of the lru cache.")
	legacyAutocommit    = flag.Bool("legacy_autocommit", false, "DEPRECATED: set this flag to true to get the legacy behavior: all transactions will need an explicit begin, and DMLs outside transactions will return an error.")
	enableForwarding    = flag.Bool("enable_forwarding", false, "if specified, this process will also expose a QueryService interface that allows other vtgates to talk through this vtgate to the underlying tablets.")
	l2vtgateAddrs       flagutil.StringListValue
	disableLocalGateway = flag.Bool("disable_local_gateway", false, "if specified, this process will not route any queries to local tablets in the local cell")
)

func getTxMode() vtgatepb.TransactionMode {
	switch *transactionMode {
	case "SINGLE":
		log.Infof("Transaction mode: '%s'", *transactionMode)
		return vtgatepb.TransactionMode_SINGLE
	case "MULTI":
		log.Infof("Transaction mode: '%s'", *transactionMode)
		return vtgatepb.TransactionMode_MULTI
	case "TWOPC":
		log.Infof("Transaction mode: '%s'", *transactionMode)
		return vtgatepb.TransactionMode_TWOPC
	default:
		log.Warningf("Unrecognized transactionMode '%s'. Continuing with default 'MULTI'", *transactionMode)
		return vtgatepb.TransactionMode_MULTI
	}
}

var (
	rpcVTGate *VTGate

	qpsByOperation *stats.Rates
	qpsByKeyspace  *stats.Rates
	qpsByDbType    *stats.Rates

	vschemaCounters *stats.Counters

	errorsByOperation *stats.Rates
	errorsByKeyspace  *stats.Rates
	errorsByDbType    *stats.Rates
	errorsByCode      *stats.Rates

	// Error counters should be global so they can be set from anywhere
	errorCounts *stats.MultiCounters

	warnings *stats.Counters
)

// VTGate is the rpc interface to vtgate. Only one instance
// can be created. It implements vtgateservice.VTGateService
// VTGate exposes multiple generations of interfaces. The V3
// interface is the latest one, which is capable of processing
// queries with no additional hints. V2 functions require
// the keyspace id or keyrange to be specified. V1 functions
// require shard info. V0 functions are informational that
// return topo information. Often, 'V2' or 'legacy' is used
// to refer to all legacy versions of the API (V2, V1 and V0).
type VTGate struct {
	// Dependency: executor->resolver->scatterConn->txConn->gateway.
	// VTGate still needs resolver and txConn to support legacy functions.
	executor *Executor
	resolver *Resolver
	txConn   *TxConn
	gw       gateway.Gateway
	l2vtgate *L2VTGate

	// stats objects.
	// TODO(sougou): This needs to be cleaned up. There
	// are global vars that depend on this member var.
	timings      *stats.MultiTimings
	rowsReturned *stats.MultiCounters

	// the throttled loggers for all errors, one per API entry
	logExecute                  *logutil.ThrottledLogger
	logStreamExecute            *logutil.ThrottledLogger
	logExecuteShards            *logutil.ThrottledLogger
	logExecuteKeyspaceIds       *logutil.ThrottledLogger
	logExecuteKeyRanges         *logutil.ThrottledLogger
	logExecuteEntityIds         *logutil.ThrottledLogger
	logExecuteBatchShards       *logutil.ThrottledLogger
	logExecuteBatchKeyspaceIds  *logutil.ThrottledLogger
	logStreamExecuteKeyspaceIds *logutil.ThrottledLogger
	logStreamExecuteKeyRanges   *logutil.ThrottledLogger
	logStreamExecuteShards      *logutil.ThrottledLogger
	logUpdateStream             *logutil.ThrottledLogger
	logMessageStream            *logutil.ThrottledLogger
}

// RegisterVTGate defines the type of registration mechanism.
type RegisterVTGate func(vtgateservice.VTGateService)

// RegisterVTGates stores register funcs for VTGate server.
var RegisterVTGates []RegisterVTGate

// Init initializes VTGate server.
func Init(ctx context.Context, hc discovery.HealthCheck, topoServer *topo.Server, serv srvtopo.Server, cell string, retryCount int, tabletTypesToWait []topodatapb.TabletType) *VTGate {
	if rpcVTGate != nil {
		log.Fatalf("VTGate already initialized")
	}

	// vschemaCounters needs to be initialized before planner to
	// catch the initial load stats.
	vschemaCounters = stats.NewCounters("VtgateVSchemaCounts")

	// Build objects from low to high level.
	// Start with the gateway. If we can't reach the topology service,
	// we can't go on much further, so we log.Fatal out.
	var gw gateway.Gateway
	var l2vtgate *L2VTGate
	if !*disableLocalGateway {
		gw = gateway.GetCreator()(hc, topoServer, serv, cell, retryCount)
		if err := gateway.WaitForTablets(gw, tabletTypesToWait); err != nil {
			log.Fatalf("gateway.WaitForTablets failed: %v", err)
		}

		// l2vtgate gives access to the underlying Gateway
		// from an exported QueryService interface.
		if *enableForwarding {
			l2vtgate = initL2VTGate(gw)
		}
	}

	// If we have other vtgate pools to connect to, create a
	// HybridGateway to perform the routing.
	if len(l2vtgateAddrs) > 0 {
		hgw, err := gateway.NewHybridGateway(gw, l2vtgateAddrs, retryCount)
		if err != nil {
			log.Fatalf("gateway.NewHybridGateway failed: %v", err)
		}
		hgw.RegisterStats()
		gw = hgw
	}

	// Check we have something to do.
	if gw == nil {
		log.Fatalf("'-disable_local_gateway' cannot be specified if 'l2vtgate_addrs' is also empty, otherwise this vtgate has no backend")
	}

	tc := NewTxConn(gw, getTxMode())
	// ScatterConn depends on TxConn to perform forced rollbacks.
	sc := NewScatterConn("VttabletCall", tc, gw, hc)
	srvResolver := srvtopo.NewResolver(serv, gw, cell)
	resolver := NewResolver(srvResolver, serv, cell, sc)

	rpcVTGate = &VTGate{
		executor:     NewExecutor(ctx, serv, cell, "VTGateExecutor", resolver, *normalizeQueries, *streamBufferSize, *queryPlanCacheSize, *legacyAutocommit),
		resolver:     resolver,
		txConn:       tc,
		gw:           gw,
		l2vtgate:     l2vtgate,
		timings:      stats.NewMultiTimings("VtgateApi", []string{"Operation", "Keyspace", "DbType"}),
		rowsReturned: stats.NewMultiCounters("VtgateApiRowsReturned", []string{"Operation", "Keyspace", "DbType"}),

		logExecute:                  logutil.NewThrottledLogger("Execute", 5*time.Second),
		logStreamExecute:            logutil.NewThrottledLogger("StreamExecute", 5*time.Second),
		logExecuteShards:            logutil.NewThrottledLogger("ExecuteShards", 5*time.Second),
		logExecuteKeyspaceIds:       logutil.NewThrottledLogger("ExecuteKeyspaceIds", 5*time.Second),
		logExecuteKeyRanges:         logutil.NewThrottledLogger("ExecuteKeyRanges", 5*time.Second),
		logExecuteEntityIds:         logutil.NewThrottledLogger("ExecuteEntityIds", 5*time.Second),
		logExecuteBatchShards:       logutil.NewThrottledLogger("ExecuteBatchShards", 5*time.Second),
		logExecuteBatchKeyspaceIds:  logutil.NewThrottledLogger("ExecuteBatchKeyspaceIds", 5*time.Second),
		logStreamExecuteKeyspaceIds: logutil.NewThrottledLogger("StreamExecuteKeyspaceIds", 5*time.Second),
		logStreamExecuteKeyRanges:   logutil.NewThrottledLogger("StreamExecuteKeyRanges", 5*time.Second),
		logStreamExecuteShards:      logutil.NewThrottledLogger("StreamExecuteShards", 5*time.Second),
		logUpdateStream:             logutil.NewThrottledLogger("UpdateStream", 5*time.Second),
		logMessageStream:            logutil.NewThrottledLogger("MessageStream", 5*time.Second),
	}

	errorCounts = stats.NewMultiCounters("VtgateApiErrorCounts", []string{"Operation", "Keyspace", "DbType", "Code"})

	qpsByOperation = stats.NewRates("QPSByOperation", stats.CounterForDimension(rpcVTGate.timings, "Operation"), 15, 1*time.Minute)
	qpsByKeyspace = stats.NewRates("QPSByKeyspace", stats.CounterForDimension(rpcVTGate.timings, "Keyspace"), 15, 1*time.Minute)
	qpsByDbType = stats.NewRates("QPSByDbType", stats.CounterForDimension(rpcVTGate.timings, "DbType"), 15*60/5, 5*time.Second)

	errorsByOperation = stats.NewRates("ErrorsByOperation", stats.CounterForDimension(errorCounts, "Operation"), 15, 1*time.Minute)
	errorsByKeyspace = stats.NewRates("ErrorsByKeyspace", stats.CounterForDimension(errorCounts, "Keyspace"), 15, 1*time.Minute)
	errorsByDbType = stats.NewRates("ErrorsByDbType", stats.CounterForDimension(errorCounts, "DbType"), 15, 1*time.Minute)
	errorsByCode = stats.NewRates("ErrorsByCode", stats.CounterForDimension(errorCounts, "Code"), 15, 1*time.Minute)

	warnings = stats.NewCounters("VTGateWarnings", "IgnoredSet")

	servenv.OnRun(func() {
		for _, f := range RegisterVTGates {
			f(rpcVTGate)
		}
	})
	rpcVTGate.registerDebugHealthHandler()
	err := initQueryLogger(rpcVTGate)
	if err != nil {
		log.Fatalf("error initializing query logger: %v", err)
	}

	return rpcVTGate
}

func (vtg *VTGate) registerDebugHealthHandler() {
	http.HandleFunc("/debug/health", func(w http.ResponseWriter, r *http.Request) {
		if err := acl.CheckAccessHTTP(r, acl.MONITORING); err != nil {
			acl.SendError(w, err)
			return
		}
		w.Header().Set("Content-Type", "text/plain")
		if err := vtg.IsHealthy(); err != nil {
			w.Write([]byte("not ok"))
			return
		}
		w.Write([]byte("ok"))
	})
}

// IsHealthy returns nil if server is healthy.
// Otherwise, it returns an error indicating the reason.
func (vtg *VTGate) IsHealthy() error {
	return nil
}

// Gateway returns the current gateway implementation. Mostly used for tests.
func (vtg *VTGate) Gateway() gateway.Gateway {
	return vtg.gw
}

// L2VTGate returns the L2VTGate object. Mostly used for tests.
func (vtg *VTGate) L2VTGate() *L2VTGate {
	return vtg.l2vtgate
}

// Execute executes a non-streaming query. This is a V3 function.
func (vtg *VTGate) Execute(ctx context.Context, session *vtgatepb.Session, sql string, bindVariables map[string]*querypb.BindVariable) (newSession *vtgatepb.Session, qr *sqltypes.Result, err error) {
	target := vtg.executor.ParseTarget(session.TargetString)
	statsKey := []string{"Execute", target.Keyspace, topoproto.TabletTypeLString(target.TabletType)}
	defer vtg.timings.Record(statsKey, time.Now())

	if bvErr := sqltypes.ValidateBindVariables(bindVariables); bvErr != nil {
		err = vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "%v", bvErr)
		goto handleError
	}

	qr, err = vtg.executor.Execute(ctx, "Execute", NewSafeSession(session), sql, bindVariables)
	if err == nil {
		vtg.rowsReturned.Add(statsKey, int64(len(qr.Rows)))
		return session, qr, nil
	}

handleError:
	query := map[string]interface{}{
		"Sql":           sql,
		"BindVariables": bindVariables,
		"Session":       session,
	}
	err = recordAndAnnotateError(err, statsKey, query, vtg.logExecute)
	return session, nil, err
}

// ExecuteBatch executes a batch of queries. This is a V3 function.
func (vtg *VTGate) ExecuteBatch(ctx context.Context, session *vtgatepb.Session, sqlList []string, bindVariablesList []map[string]*querypb.BindVariable) (*vtgatepb.Session, []sqltypes.QueryResponse, error) {
	target := vtg.executor.ParseTarget(session.TargetString)
	statsKey := []string{"ExecuteBatch", target.Keyspace, topoproto.TabletTypeLString(target.TabletType)}
	defer vtg.timings.Record(statsKey, time.Now())

	for _, bindVariables := range bindVariablesList {
		if bvErr := sqltypes.ValidateBindVariables(bindVariables); bvErr != nil {
			return session, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "%v", bvErr)
		}
	}

	qrl := make([]sqltypes.QueryResponse, len(sqlList))
	for i, sql := range sqlList {
		var bv map[string]*querypb.BindVariable
		if len(bindVariablesList) != 0 {
			bv = bindVariablesList[i]
		}
		session, qrl[i].QueryResult, qrl[i].QueryError = vtg.Execute(ctx, session, sql, bv)
		if qr := qrl[i].QueryResult; qr != nil {
			vtg.rowsReturned.Add(statsKey, int64(len(qr.Rows)))
		}
	}
	return session, qrl, nil
}

// StreamExecute executes a streaming query. This is a V3 function.
func (vtg *VTGate) StreamExecute(ctx context.Context, session *vtgatepb.Session, sql string, bindVariables map[string]*querypb.BindVariable, callback func(*sqltypes.Result) error) error {
	target := vtg.executor.ParseTarget(session.TargetString)
	statsKey := []string{"StreamExecute", target.Keyspace, topoproto.TabletTypeLString(target.TabletType)}
	defer vtg.timings.Record(statsKey, time.Now())

	var err error
	if bvErr := sqltypes.ValidateBindVariables(bindVariables); bvErr != nil {
		err = vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "%v", bvErr)
		goto handleError
	}

	if target.Shard != "" {
		err = vtg.resolver.streamExecute(
			ctx,
			sql,
			bindVariables,
			target.Keyspace,
			target.TabletType,
			func(keyspace string) ([]*srvtopo.ResolvedShard, error) {
				return vtg.resolver.resolver.ResolveDestination(ctx, keyspace, target.TabletType, key.DestinationShard(target.Shard))
			},
			session.Options,
			func(reply *sqltypes.Result) error {
				vtg.rowsReturned.Add(statsKey, int64(len(reply.Rows)))
				return callback(reply)
			})
	} else {
		err = vtg.executor.StreamExecute(
			ctx,
			"StreamExecute",
			NewSafeSession(session),
			sql,
			bindVariables,
			target,
			func(reply *sqltypes.Result) error {
				vtg.rowsReturned.Add(statsKey, int64(len(reply.Rows)))
				return callback(reply)
			})
	}

handleError:
	if err != nil {
		query := map[string]interface{}{
			"Sql":           sql,
			"BindVariables": bindVariables,
			"Session":       session,
		}
		return recordAndAnnotateError(err, statsKey, query, vtg.logStreamExecute)
	}
	return nil
}

// ExecuteShards executes a non-streaming query on the specified shards.
// This is a legacy function.
func (vtg *VTGate) ExecuteShards(ctx context.Context, sql string, bindVariables map[string]*querypb.BindVariable, keyspace string, shards []string, tabletType topodatapb.TabletType, session *vtgatepb.Session, notInTransaction bool, options *querypb.ExecuteOptions) (*sqltypes.Result, error) {
	startTime := time.Now()
	ltt := topoproto.TabletTypeLString(tabletType)
	statsKey := []string{"ExecuteShards", keyspace, ltt}
	defer vtg.timings.Record(statsKey, startTime)

	var qr *sqltypes.Result
	var err error

	if bvErr := sqltypes.ValidateBindVariables(bindVariables); bvErr != nil {
		err = vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "%v", bvErr)
		goto handleError
	}

	sql = sqlannotation.AnnotateIfDML(sql, nil)

	qr, err = vtg.resolver.Execute(
		ctx,
		sql,
		bindVariables,
		tabletType,
		session,
		func() ([]*srvtopo.ResolvedShard, error) {
			return vtg.resolver.resolver.ResolveDestination(ctx, keyspace, tabletType, key.DestinationShards(shards))
		},
		notInTransaction,
		options,
		nil,
	)
	if err == nil {
		vtg.rowsReturned.Add(statsKey, int64(len(qr.Rows)))
		return qr, nil
	}

handleError:
	query := map[string]interface{}{
		"Sql":              sql,
		"BindVariables":    bindVariables,
		"Keyspace":         keyspace,
		"Shards":           shards,
		"TabletType":       ltt,
		"Session":          session,
		"NotInTransaction": notInTransaction,
		"Options":          options,
	}
	err = recordAndAnnotateError(err, statsKey, query, vtg.logExecuteShards)
	return nil, err
}

// ExecuteKeyspaceIds executes a non-streaming query based on the specified keyspace ids. This is a legacy function.
func (vtg *VTGate) ExecuteKeyspaceIds(ctx context.Context, sql string, bindVariables map[string]*querypb.BindVariable, keyspace string, keyspaceIds [][]byte, tabletType topodatapb.TabletType, session *vtgatepb.Session, notInTransaction bool, options *querypb.ExecuteOptions) (*sqltypes.Result, error) {
	startTime := time.Now()
	ltt := topoproto.TabletTypeLString(tabletType)
	statsKey := []string{"ExecuteKeyspaceIds", keyspace, ltt}
	defer vtg.timings.Record(statsKey, startTime)

	var qr *sqltypes.Result
	var err error

	if bvErr := sqltypes.ValidateBindVariables(bindVariables); bvErr != nil {
		err = vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "%v", bvErr)
		goto handleError
	}

	sql = sqlannotation.AnnotateIfDML(sql, keyspaceIds)

	qr, err = vtg.resolver.ExecuteKeyspaceIds(ctx, sql, bindVariables, keyspace, keyspaceIds, tabletType, session, notInTransaction, options)
	if err == nil {
		vtg.rowsReturned.Add(statsKey, int64(len(qr.Rows)))
		return qr, nil
	}

handleError:
	query := map[string]interface{}{
		"Sql":              sql,
		"BindVariables":    bindVariables,
		"Keyspace":         keyspace,
		"KeyspaceIds":      keyspaceIds,
		"TabletType":       ltt,
		"Session":          session,
		"NotInTransaction": notInTransaction,
		"Options":          options,
	}
	err = recordAndAnnotateError(err, statsKey, query, vtg.logExecuteKeyspaceIds)
	return nil, err
}

// ExecuteKeyRanges executes a non-streaming query based on the specified keyranges. This is a legacy function.
func (vtg *VTGate) ExecuteKeyRanges(ctx context.Context, sql string, bindVariables map[string]*querypb.BindVariable, keyspace string, keyRanges []*topodatapb.KeyRange, tabletType topodatapb.TabletType, session *vtgatepb.Session, notInTransaction bool, options *querypb.ExecuteOptions) (*sqltypes.Result, error) {
	startTime := time.Now()
	ltt := topoproto.TabletTypeLString(tabletType)
	statsKey := []string{"ExecuteKeyRanges", keyspace, ltt}
	defer vtg.timings.Record(statsKey, startTime)

	var qr *sqltypes.Result
	var err error

	if bvErr := sqltypes.ValidateBindVariables(bindVariables); bvErr != nil {
		err = vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "%v", bvErr)
		goto handleError
	}

	sql = sqlannotation.AnnotateIfDML(sql, nil)

	qr, err = vtg.resolver.ExecuteKeyRanges(ctx, sql, bindVariables, keyspace, keyRanges, tabletType, session, notInTransaction, options)
	if err == nil {
		vtg.rowsReturned.Add(statsKey, int64(len(qr.Rows)))
		return qr, nil
	}

handleError:
	query := map[string]interface{}{
		"Sql":              sql,
		"BindVariables":    bindVariables,
		"Keyspace":         keyspace,
		"KeyRanges":        keyRanges,
		"TabletType":       ltt,
		"Session":          session,
		"NotInTransaction": notInTransaction,
		"Options":          options,
	}
	err = recordAndAnnotateError(err, statsKey, query, vtg.logExecuteKeyRanges)
	return nil, err
}

// ExecuteEntityIds excutes a non-streaming query based on given KeyspaceId map. This is a legacy function.
func (vtg *VTGate) ExecuteEntityIds(ctx context.Context, sql string, bindVariables map[string]*querypb.BindVariable, keyspace string, entityColumnName string, entityKeyspaceIDs []*vtgatepb.ExecuteEntityIdsRequest_EntityId, tabletType topodatapb.TabletType, session *vtgatepb.Session, notInTransaction bool, options *querypb.ExecuteOptions) (*sqltypes.Result, error) {
	startTime := time.Now()
	ltt := topoproto.TabletTypeLString(tabletType)
	statsKey := []string{"ExecuteEntityIds", keyspace, ltt}
	defer vtg.timings.Record(statsKey, startTime)

	var qr *sqltypes.Result
	var err error

	if bvErr := sqltypes.ValidateBindVariables(bindVariables); bvErr != nil {
		err = vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "%v", bvErr)
		goto handleError
	}

	sql = sqlannotation.AnnotateIfDML(sql, nil)

	qr, err = vtg.resolver.ExecuteEntityIds(ctx, sql, bindVariables, keyspace, entityColumnName, entityKeyspaceIDs, tabletType, session, notInTransaction, options)
	if err == nil {
		vtg.rowsReturned.Add(statsKey, int64(len(qr.Rows)))
		return qr, nil
	}

handleError:
	query := map[string]interface{}{
		"Sql":               sql,
		"BindVariables":     bindVariables,
		"Keyspace":          keyspace,
		"EntityColumnName":  entityColumnName,
		"EntityKeyspaceIDs": entityKeyspaceIDs,
		"TabletType":        ltt,
		"Session":           session,
		"NotInTransaction":  notInTransaction,
		"Options":           options,
	}
	err = recordAndAnnotateError(err, statsKey, query, vtg.logExecuteEntityIds)
	return nil, err
}

// ExecuteBatchShards executes a group of queries on the specified shards. This is a legacy function.
func (vtg *VTGate) ExecuteBatchShards(ctx context.Context, queries []*vtgatepb.BoundShardQuery, tabletType topodatapb.TabletType, asTransaction bool, session *vtgatepb.Session, options *querypb.ExecuteOptions) ([]sqltypes.Result, error) {
	startTime := time.Now()
	ltt := topoproto.TabletTypeLString(tabletType)
	statsKey := []string{"ExecuteBatchShards", unambiguousKeyspaceBSQ(queries), ltt}
	defer vtg.timings.Record(statsKey, startTime)

	var qrs []sqltypes.Result
	var err error

	for _, query := range queries {
		if bvErr := sqltypes.ValidateBindVariables(query.Query.BindVariables); bvErr != nil {
			err = vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "%v", bvErr)
			goto handleError
		}
	}

	annotateBoundShardQueriesAsUnfriendly(queries)

	qrs, err = vtg.resolver.ExecuteBatch(
		ctx,
		tabletType,
		asTransaction,
		session,
		options,
		func() (*scatterBatchRequest, error) {
			return boundShardQueriesToScatterBatchRequest(ctx, vtg.resolver.resolver, queries, tabletType)
		})
	if err == nil {
		var rowCount int64
		for _, qr := range qrs {
			rowCount += int64(len(qr.Rows))
		}
		vtg.rowsReturned.Add(statsKey, rowCount)
		return qrs, nil
	}

handleError:
	query := map[string]interface{}{
		"Queries":       queries,
		"TabletType":    ltt,
		"AsTransaction": asTransaction,
		"Session":       session,
		"Options":       options,
	}
	err = recordAndAnnotateError(err, statsKey, query, vtg.logExecuteBatchShards)
	return nil, err
}

// ExecuteBatchKeyspaceIds executes a group of queries based on the specified keyspace ids. This is a legacy function.
func (vtg *VTGate) ExecuteBatchKeyspaceIds(ctx context.Context, queries []*vtgatepb.BoundKeyspaceIdQuery, tabletType topodatapb.TabletType, asTransaction bool, session *vtgatepb.Session, options *querypb.ExecuteOptions) ([]sqltypes.Result, error) {
	startTime := time.Now()
	ltt := topoproto.TabletTypeLString(tabletType)
	statsKey := []string{"ExecuteBatchKeyspaceIds", unambiguousKeyspaceBKSIQ(queries), ltt}
	defer vtg.timings.Record(statsKey, startTime)

	var qrs []sqltypes.Result
	var err error

	for _, query := range queries {
		if bvErr := sqltypes.ValidateBindVariables(query.Query.BindVariables); bvErr != nil {
			err = vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "%v", bvErr)
			goto handleError
		}
	}

	annotateBoundKeyspaceIDQueries(queries)

	qrs, err = vtg.resolver.ExecuteBatch(
		ctx,
		tabletType,
		asTransaction,
		session,
		options,
		func() (*scatterBatchRequest, error) {
			return boundKeyspaceIDQueriesToScatterBatchRequest(ctx, vtg.resolver.resolver, queries, tabletType)
		})
	if err == nil {
		var rowCount int64
		for _, qr := range qrs {
			rowCount += int64(len(qr.Rows))
		}
		vtg.rowsReturned.Add(statsKey, rowCount)
		return qrs, nil
	}

handleError:
	query := map[string]interface{}{
		"Queries":       queries,
		"TabletType":    ltt,
		"AsTransaction": asTransaction,
		"Session":       session,
		"Options":       options,
	}
	err = recordAndAnnotateError(err, statsKey, query, vtg.logExecuteBatchKeyspaceIds)
	return nil, err
}

// StreamExecuteKeyspaceIds executes a streaming query on the specified KeyspaceIds.
// The KeyspaceIds are resolved to shards using the serving graph.
// This function currently temporarily enforces the restriction of executing on
// one shard since it cannot merge-sort the results to guarantee ordering of
// response which is needed for checkpointing.
// The api supports supplying multiple KeyspaceIds to make it future proof. This is a legacy function.
func (vtg *VTGate) StreamExecuteKeyspaceIds(ctx context.Context, sql string, bindVariables map[string]*querypb.BindVariable, keyspace string, keyspaceIds [][]byte, tabletType topodatapb.TabletType, options *querypb.ExecuteOptions, callback func(*sqltypes.Result) error) error {
	startTime := time.Now()
	ltt := topoproto.TabletTypeLString(tabletType)
	statsKey := []string{"StreamExecuteKeyspaceIds", keyspace, ltt}
	defer vtg.timings.Record(statsKey, startTime)

	var err error

	if bvErr := sqltypes.ValidateBindVariables(bindVariables); bvErr != nil {
		err = vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "%v", bvErr)
		goto handleError
	}

	err = vtg.resolver.StreamExecuteKeyspaceIds(
		ctx,
		sql,
		bindVariables,
		keyspace,
		keyspaceIds,
		tabletType,
		options,
		func(reply *sqltypes.Result) error {
			vtg.rowsReturned.Add(statsKey, int64(len(reply.Rows)))
			return callback(reply)
		})

handleError:
	if err != nil {
		query := map[string]interface{}{
			"Sql":           sql,
			"BindVariables": bindVariables,
			"Keyspace":      keyspace,
			"KeyspaceIds":   keyspaceIds,
			"TabletType":    ltt,
			"Options":       options,
		}
		return recordAndAnnotateError(err, statsKey, query, vtg.logStreamExecuteKeyspaceIds)
	}
	return nil
}

// StreamExecuteKeyRanges executes a streaming query on the specified KeyRanges.
// The KeyRanges are resolved to shards using the serving graph.
// This function currently temporarily enforces the restriction of executing on
// one shard since it cannot merge-sort the results to guarantee ordering of
// response which is needed for checkpointing.
// The api supports supplying multiple keyranges to make it future proof. This is a legacy function.
func (vtg *VTGate) StreamExecuteKeyRanges(ctx context.Context, sql string, bindVariables map[string]*querypb.BindVariable, keyspace string, keyRanges []*topodatapb.KeyRange, tabletType topodatapb.TabletType, options *querypb.ExecuteOptions, callback func(*sqltypes.Result) error) error {
	startTime := time.Now()
	ltt := topoproto.TabletTypeLString(tabletType)
	statsKey := []string{"StreamExecuteKeyRanges", keyspace, ltt}
	defer vtg.timings.Record(statsKey, startTime)

	var err error

	if bvErr := sqltypes.ValidateBindVariables(bindVariables); bvErr != nil {
		err = vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "%v", bvErr)
		goto handleError
	}

	err = vtg.resolver.StreamExecuteKeyRanges(
		ctx,
		sql,
		bindVariables,
		keyspace,
		keyRanges,
		tabletType,
		options,
		func(reply *sqltypes.Result) error {
			vtg.rowsReturned.Add(statsKey, int64(len(reply.Rows)))
			return callback(reply)
		})

handleError:
	if err != nil {
		query := map[string]interface{}{
			"Sql":           sql,
			"BindVariables": bindVariables,
			"Keyspace":      keyspace,
			"KeyRanges":     keyRanges,
			"TabletType":    ltt,
			"Options":       options,
		}
		return recordAndAnnotateError(err, statsKey, query, vtg.logStreamExecuteKeyRanges)
	}
	return nil
}

// StreamExecuteShards executes a streaming query on the specified shards. This is a legacy function.
func (vtg *VTGate) StreamExecuteShards(ctx context.Context, sql string, bindVariables map[string]*querypb.BindVariable, keyspace string, shards []string, tabletType topodatapb.TabletType, options *querypb.ExecuteOptions, callback func(*sqltypes.Result) error) error {
	startTime := time.Now()
	ltt := topoproto.TabletTypeLString(tabletType)
	statsKey := []string{"StreamExecuteShards", keyspace, ltt}
	defer vtg.timings.Record(statsKey, startTime)

	var err error

	if bvErr := sqltypes.ValidateBindVariables(bindVariables); bvErr != nil {
		err = vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "%v", bvErr)
		goto handleError
	}

	err = vtg.resolver.streamExecute(
		ctx,
		sql,
		bindVariables,
		keyspace,
		tabletType,
		func(keyspace string) ([]*srvtopo.ResolvedShard, error) {
			return vtg.resolver.resolver.ResolveDestination(ctx, keyspace, tabletType, key.DestinationShards(shards))
		},
		options,
		func(reply *sqltypes.Result) error {
			vtg.rowsReturned.Add(statsKey, int64(len(reply.Rows)))
			return callback(reply)
		})

handleError:
	if err != nil {
		query := map[string]interface{}{
			"Sql":           sql,
			"BindVariables": bindVariables,
			"Keyspace":      keyspace,
			"Shards":        shards,
			"TabletType":    ltt,
			"Options":       options,
		}
		return recordAndAnnotateError(err, statsKey, query, vtg.logStreamExecuteShards)
	}
	return nil
}

// Begin begins a transaction. This is a legacy function.
func (vtg *VTGate) Begin(ctx context.Context, singledb bool) (*vtgatepb.Session, error) {
	if !singledb && vtg.txConn.mode == vtgatepb.TransactionMode_SINGLE {
		return nil, vterrors.New(vtrpcpb.Code_INVALID_ARGUMENT, "multi-db transaction disallowed")
	}
	return &vtgatepb.Session{
		InTransaction: true,
		SingleDb:      singledb,
	}, nil
}

// Commit commits a transaction. This is a legacy function.
func (vtg *VTGate) Commit(ctx context.Context, twopc bool, session *vtgatepb.Session) error {
	if session == nil {
		return formatError(vterrors.New(vtrpcpb.Code_INVALID_ARGUMENT, "cannot commit: empty session"))
	}
	if !session.InTransaction {
		return formatError(vterrors.New(vtrpcpb.Code_ABORTED, "cannot commit: not in transaction"))
	}
	if twopc {
		session.TransactionMode = vtgatepb.TransactionMode_TWOPC
	}
	return formatError(vtg.txConn.Commit(ctx, NewSafeSession(session)))
}

// Rollback rolls back a transaction. This is a legacy function.
func (vtg *VTGate) Rollback(ctx context.Context, session *vtgatepb.Session) error {
	return formatError(vtg.txConn.Rollback(ctx, NewSafeSession(session)))
}

// ResolveTransaction resolves the specified 2PC transaction.
func (vtg *VTGate) ResolveTransaction(ctx context.Context, dtid string) error {
	return formatError(vtg.txConn.Resolve(ctx, dtid))
}

// isKeyspaceRangeBasedSharded returns true if a keyspace is sharded
// by range.  This is true when there is a ShardingColumnType defined
// in the SrvKeyspace (that is using the range-based sharding with the
// client specifying the sharding key), or when the VSchema for the
// keyspace is Sharded.
func (vtg *VTGate) isKeyspaceRangeBasedSharded(keyspace string, srvKeyspace *topodatapb.SrvKeyspace) bool {
	if srvKeyspace.ShardingColumnType != topodatapb.KeyspaceIdType_UNSET {
		// We are using range based sharding with the application
		// providing the sharding key value.
		return true
	}
	if vtg.executor.IsKeyspaceRangeBasedSharded(keyspace) {
		// We are using range based sharding with the VSchema
		// poviding the routing information
		return true
	}

	// Not range based sharded, might be un-sharded or custom sharded.
	return false
}

// SplitQuery implements the SplitQuery RPC. This is the new version that
// supports multiple split-columns and multiple splitting algorithms.
// See the documentation of SplitQueryRequest in "proto/vtgate.proto" for more
// information.
func (vtg *VTGate) SplitQuery(
	ctx context.Context,
	keyspace string,
	sql string,
	bindVariables map[string]*querypb.BindVariable,
	splitColumns []string,
	splitCount int64,
	numRowsPerQueryPart int64,
	algorithm querypb.SplitQueryRequest_Algorithm) ([]*vtgatepb.SplitQueryResponse_Part, error) {

	if bvErr := sqltypes.ValidateBindVariables(bindVariables); bvErr != nil {
		return nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "%v", bvErr)
	}

	// TODO(erez): Add validation of SplitQuery parameters.
	rss, srvKeyspace, err := vtg.resolver.resolver.GetAllShards(ctx, keyspace, topodatapb.TabletType_RDONLY)
	if err != nil {
		return nil, err
	}

	// If the caller specified a splitCount (vs. specifying 'numRowsPerQueryPart') scale it by the
	// number of shards (otherwise it stays 0).
	perShardSplitCount := int64(math.Ceil(float64(splitCount) / float64(len(rss))))

	// Determine whether to return SplitQueryResponse_KeyRangeParts or SplitQueryResponse_ShardParts.
	// We return 'KeyRangeParts' for sharded keyspaces that are not custom sharded. If the
	// keyspace is custom sharded or unsharded we return 'ShardParts'.
	var querySplitToQueryPartFunc func(
		querySplit *querypb.QuerySplit, rs *srvtopo.ResolvedShard) (*vtgatepb.SplitQueryResponse_Part, error)
	if vtg.isKeyspaceRangeBasedSharded(keyspace, srvKeyspace) {
		querySplitToQueryPartFunc = func(querySplit *querypb.QuerySplit, rs *srvtopo.ResolvedShard) (*vtgatepb.SplitQueryResponse_Part, error) {
			// Use ValidateShardName to extract the keyrange.
			_, kr, err := topo.ValidateShardName(rs.Target.Shard)
			if err != nil {
				return nil, fmt.Errorf("cannot extract keyrange from shard name %v: %v", rs.Target.Shard, err)
			}
			if kr == nil {
				// Keyrange can be nil for the shard (e.g. for single-sharded keyspaces during resharding).
				// In this case we append an empty keyrange that represents the entire keyspace.
				kr = &topodatapb.KeyRange{
					Start: []byte{},
					End:   []byte{},
				}
			}
			return &vtgatepb.SplitQueryResponse_Part{
				Query: querySplit.Query,
				KeyRangePart: &vtgatepb.SplitQueryResponse_KeyRangePart{
					Keyspace:  keyspace,
					KeyRanges: []*topodatapb.KeyRange{kr},
				},
				Size: querySplit.RowCount,
			}, nil
		}
	} else {
		// Keyspace is either unsharded or custom-sharded.
		querySplitToQueryPartFunc = func(querySplit *querypb.QuerySplit, rs *srvtopo.ResolvedShard) (*vtgatepb.SplitQueryResponse_Part, error) {
			return &vtgatepb.SplitQueryResponse_Part{
				Query: querySplit.Query,
				ShardPart: &vtgatepb.SplitQueryResponse_ShardPart{
					Keyspace: keyspace,
					Shards:   []string{rs.Target.Shard},
				},
				Size: querySplit.RowCount,
			}, nil
		}
	}

	return vtg.resolver.scatterConn.SplitQuery(
		ctx,
		sql,
		bindVariables,
		splitColumns,
		perShardSplitCount,
		numRowsPerQueryPart,
		algorithm,
		rss,
		querySplitToQueryPartFunc)
}

// GetSrvKeyspace is part of the vtgate service API.
func (vtg *VTGate) GetSrvKeyspace(ctx context.Context, keyspace string) (*topodatapb.SrvKeyspace, error) {
	return vtg.resolver.toposerv.GetSrvKeyspace(ctx, vtg.resolver.cell, keyspace)
}

// MessageStream is part of the vtgate service API. This is a V2 level API that's sent
// to the Resolver.
func (vtg *VTGate) MessageStream(ctx context.Context, keyspace string, shard string, keyRange *topodatapb.KeyRange, name string, callback func(*sqltypes.Result) error) error {
	startTime := time.Now()
	ltt := topoproto.TabletTypeLString(topodatapb.TabletType_MASTER)
	statsKey := []string{"MessageStream", keyspace, ltt}
	defer vtg.timings.Record(statsKey, startTime)

	err := vtg.executor.MessageStream(
		ctx,
		keyspace,
		shard,
		keyRange,
		name,
		callback,
	)
	if err != nil {
		request := map[string]interface{}{
			"Keyspace":    keyspace,
			"Shard":       shard,
			"KeyRange":    keyRange,
			"TabletType":  ltt,
			"MessageName": name,
		}
		recordAndAnnotateError(err, statsKey, request, vtg.logMessageStream)
	}
	return formatError(err)
}

// MessageAck is part of the vtgate service API. This is a V3 level API that's sent
// to the executor. The table name will be resolved using V3 rules, and the routing
// will make use of vindexes for sharded keyspaces.
// TODO(sougou): Deprecate this in favor of an SQL statement.
func (vtg *VTGate) MessageAck(ctx context.Context, keyspace string, name string, ids []*querypb.Value) (int64, error) {
	startTime := time.Now()
	ltt := topoproto.TabletTypeLString(topodatapb.TabletType_MASTER)
	statsKey := []string{"MessageAck", keyspace, ltt}
	defer vtg.timings.Record(statsKey, startTime)

	for _, id := range ids {
		if _, err := sqltypes.NewValue(id.Type, id.Value); err != nil {
			return 0, formatError(err)
		}
	}

	count, err := vtg.executor.MessageAck(ctx, keyspace, name, ids)
	return count, formatError(err)
}

// MessageAckKeyspaceIds is part of the vtgate service API. It routes
// message acks based on the associated keyspace ids.
func (vtg *VTGate) MessageAckKeyspaceIds(ctx context.Context, keyspace string, name string, idKeyspaceIDs []*vtgatepb.IdKeyspaceId) (int64, error) {
	startTime := time.Now()
	ltt := topoproto.TabletTypeLString(topodatapb.TabletType_MASTER)
	statsKey := []string{"MessageAckKeyspaceIds", keyspace, ltt}
	defer vtg.timings.Record(statsKey, startTime)

	for _, idKeyspaceID := range idKeyspaceIDs {
		if _, err := sqltypes.NewValue(idKeyspaceID.Id.Type, idKeyspaceID.Id.Value); err != nil {
			return 0, formatError(err)
		}
	}

	count, err := vtg.resolver.MessageAckKeyspaceIds(ctx, keyspace, name, idKeyspaceIDs)
	return count, formatError(err)
}

// UpdateStream is part of the vtgate service API.
func (vtg *VTGate) UpdateStream(ctx context.Context, keyspace string, shard string, keyRange *topodatapb.KeyRange, tabletType topodatapb.TabletType, timestamp int64, event *querypb.EventToken, callback func(*querypb.StreamEvent, int64) error) error {
	startTime := time.Now()
	ltt := topoproto.TabletTypeLString(tabletType)
	statsKey := []string{"UpdateStream", keyspace, ltt}
	defer vtg.timings.Record(statsKey, startTime)

	err := vtg.resolver.UpdateStream(
		ctx,
		keyspace,
		shard,
		keyRange,
		tabletType,
		timestamp,
		event,
		callback,
	)
	if err != nil {
		request := map[string]interface{}{
			"Keyspace":   keyspace,
			"Shard":      shard,
			"KeyRange":   keyRange,
			"TabletType": ltt,
			"Timestamp":  timestamp,
		}
		recordAndAnnotateError(err, statsKey, request, vtg.logUpdateStream)
	}
	return formatError(err)
}

// GetGatewayCacheStatus returns a displayable version of the Gateway cache.
func (vtg *VTGate) GetGatewayCacheStatus() gateway.TabletCacheStatusList {
	return vtg.resolver.GetGatewayCacheStatus()
}

// VSchemaStats returns the loaded vschema stats.
func (vtg *VTGate) VSchemaStats() *VSchemaStats {
	return vtg.executor.VSchemaStats()
}

func truncateErrorStrings(data map[string]interface{}) map[string]interface{} {
	ret := map[string]interface{}{}
	for key, val := range data {
		mapVal, ok := val.(map[string]interface{})
		if ok {
			ret[key] = truncateErrorStrings(mapVal)
		} else {
			strVal := fmt.Sprintf("%v", val)
			ret[key] = sqlparser.TruncateForLog(strVal)
		}
	}
	return ret
}

func recordAndAnnotateError(err error, statsKey []string, request map[string]interface{}, logger *logutil.ThrottledLogger) error {
	ec := vterrors.Code(err)
	fullKey := []string{
		statsKey[0],
		statsKey[1],
		statsKey[2],
		ec.String(),
	}

	// Traverse the request structure and truncate any long values
	request = truncateErrorStrings(request)

	errorCounts.Add(fullKey, 1)
	// Most errors are not logged by vtgate because they're either too spammy or logged elsewhere.
	switch ec {
	case vtrpcpb.Code_UNKNOWN, vtrpcpb.Code_INTERNAL, vtrpcpb.Code_DATA_LOSS:
		logger.Errorf("%v, request: %+v", err, request)
	case vtrpcpb.Code_UNAVAILABLE:
		logger.Infof("%v, request: %+v", err, request)
	}
	return vterrors.Wrapf(err, "vtgate: %s", servenv.ListeningURL.String())
}

func formatError(err error) error {
	if err == nil {
		return nil
	}
	return vterrors.Wrapf(err, "vtgate: %s", servenv.ListeningURL.String())
}

// HandlePanic recovers from panics, and logs / increment counters
func (vtg *VTGate) HandlePanic(err *error) {
	if x := recover(); x != nil {
		log.Errorf("Uncaught panic:\n%v\n%s", x, tb.Stack(4))
		*err = fmt.Errorf("uncaught panic: %v, vtgate: %v", x, servenv.ListeningURL.String())
		errorCounts.Add([]string{"Panic", "Unknown", "Unknown", vtrpcpb.Code_INTERNAL.String()}, 1)
	}
}

// Helper function used in ExecuteBatchKeyspaceIds
func annotateBoundKeyspaceIDQueries(queries []*vtgatepb.BoundKeyspaceIdQuery) {
	for i, q := range queries {
		queries[i].Query.Sql = sqlannotation.AnnotateIfDML(q.Query.Sql, q.KeyspaceIds)
	}
}

// Helper function used in ExecuteBatchShards
func annotateBoundShardQueriesAsUnfriendly(queries []*vtgatepb.BoundShardQuery) {
	for i, q := range queries {
		queries[i].Query.Sql = sqlannotation.AnnotateIfDML(q.Query.Sql, nil)
	}
}

// unambiguousKeyspaceBKSIQ is a helper function used in the
// ExecuteBatchKeyspaceIds method to determine the "keyspace" label for the
// stats reporting.
// If all queries target the same keyspace, it returns that keyspace.
// Otherwise it returns an empty string.
func unambiguousKeyspaceBKSIQ(queries []*vtgatepb.BoundKeyspaceIdQuery) string {
	switch len(queries) {
	case 0:
		return ""
	case 1:
		return queries[0].Keyspace
	default:
		keyspace := queries[0].Keyspace
		for _, q := range queries[1:] {
			if q.Keyspace != keyspace {
				// Request targets at least two different keyspaces.
				return ""
			}
		}
		return keyspace
	}
}

// unambiguousKeyspaceBSQ is the same as unambiguousKeyspaceBKSIQ but for the
// ExecuteBatchShards method. We are intentionally duplicating the code here and
// do not try to generalize it because this may be less performant.
func unambiguousKeyspaceBSQ(queries []*vtgatepb.BoundShardQuery) string {
	switch len(queries) {
	case 0:
		return ""
	case 1:
		return queries[0].Keyspace
	default:
		keyspace := queries[0].Keyspace
		for _, q := range queries[1:] {
			if q.Keyspace != keyspace {
				// Request targets at least two different keyspaces.
				return ""
			}
		}
		return keyspace
	}
}

func init() {
	flag.Var(&l2vtgateAddrs, "l2vtgate_addrs", "Specifies a comma-separated list of other l2 vtgate pools to connect to. These other vtgates must run with the --enable_forwarding flag")
}