text:
//go:build integration
// +build integration

package integration

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"net/url"
	"os"
	"os/exec"
	"path/filepath"
	"reflect"
	"strings"
	"testing"
	"time"

	"github.com/blang/semver/v4"
	retryablehttp "github.com/hashicorp/go-retryablehttp"
	"k8s.io/minikube/pkg/kapi"
	"k8s.io/minikube/pkg/minikube/detect"
	"k8s.io/minikube/pkg/util/retry"
)

// TestAddons tests addons that require no special environment in parallel
func TestAddons(t *testing.T) {
	profile := UniqueProfileName("addons")
	ctx, cancel := context.WithTimeout(context.Background(), Minutes(40))
	defer Cleanup(t, profile, cancel)

	setupSucceeded := t.Run("Setup", func(t *testing.T) {
		// Set an env var to point to our dummy credentials file
		// don't use t.Setenv because we sometimes manually unset the env var later manually
		err := os.Setenv("GOOGLE_APPLICATION_CREDENTIALS", filepath.Join(*testdataDir, "gcp-creds.json"))
		t.Cleanup(func() {
			os.Unsetenv("GOOGLE_APPLICATION_CREDENTIALS")
		})
		if err != nil {
			t.Fatalf("Failed setting GOOGLE_APPLICATION_CREDENTIALS env var: %v", err)
		}

		err = os.Setenv("GOOGLE_CLOUD_PROJECT", "this_is_fake")
		t.Cleanup(func() {
			os.Unsetenv("GOOGLE_CLOUD_PROJECT")
		})
		if err != nil {
			t.Fatalf("Failed setting GOOGLE_CLOUD_PROJECT env var: %v", err)
		}

		args := append([]string{"start", "-p", profile, "--wait=true", "--memory=4000", "--alsologtostderr", "--addons=registry", "--addons=metrics-server", "--addons=olm", "--addons=volumesnapshots", "--addons=csi-hostpath-driver", "--addons=gcp-auth"}, StartArgs()...)
		if !NoneDriver() { // none driver does not support ingress
			args = append(args, "--addons=ingress", "--addons=ingress-dns")
		}
		if !arm64Platform() {
			args = append(args, "--addons=helm-tiller")
		}
		rr, err := Run(t, exec.CommandContext(ctx, Target(), args...))
		if err != nil {
			t.Fatalf("%s failed: %v", rr.Command(), err)
		}

	})

	if !setupSucceeded {
		t.Fatalf("Failed setup for addon tests")
	}

	// Parallelized tests
	t.Run("parallel", func(t *testing.T) {
		tests := []struct {
			name      string
			validator validateFunc
		}{
			{"Registry", validateRegistryAddon},
			{"Ingress", validateIngressAddon},
			{"MetricsServer", validateMetricsServerAddon},
			{"HelmTiller", validateHelmTillerAddon},
			{"Olm", validateOlmAddon},
			{"CSI", validateCSIDriverAndSnapshots},
		}
		for _, tc := range tests {
			tc := tc
			if ctx.Err() == context.DeadlineExceeded {
				t.Fatalf("Unable to run more tests (deadline exceeded)")
			}
			t.Run(tc.name, func(t *testing.T) {
				MaybeParallel(t)
				tc.validator(ctx, t, profile)
			})
		}
	})

	// Run other tests after to avoid collision
	t.Run("serial", func(t *testing.T) {
		tests := []struct {
			name      string
			validator validateFunc
		}{
			{"GCPAuth", validateGCPAuthAddon},
		}
		for _, tc := range tests {
			tc := tc
			if ctx.Err() == context.DeadlineExceeded {
				t.Fatalf("Unable to run more tests (deadline exceeded)")
			}
			t.Run(tc.name, func(t *testing.T) {
				tc.validator(ctx, t, profile)
			})
		}
	})

	t.Run("StoppedEnableDisable", func(t *testing.T) {
		// Assert that disable/enable works offline
		rr, err := Run(t, exec.CommandContext(ctx, Target(), "stop", "-p", profile))
		if err != nil {
			t.Errorf("failed to stop minikube. args %q : %v", rr.Command(), err)
		}
		rr, err = Run(t, exec.CommandContext(ctx, Target(), "addons", "enable", "dashboard", "-p", profile))
		if err != nil {
			t.Errorf("failed to enable dashboard addon: args %q : %v", rr.Command(), err)
		}
		rr, err = Run(t, exec.CommandContext(ctx, Target(), "addons", "disable", "dashboard", "-p", profile))
		if err != nil {
			t.Errorf("failed to disable dashboard addon: args %q : %v", rr.Command(), err)
		}
	})
}

// validateIngressAddon tests the ingress addon by deploying a default nginx pod
func validateIngressAddon(ctx context.Context, t *testing.T, profile string) {
	defer PostMortemLogs(t, profile)
	if NoneDriver() {
		t.Skipf("skipping: ingress not supported")
	}

	client, err := kapi.Client(profile)
	if err != nil {
		t.Fatalf("failed to get Kubernetes client: %v", client)
	}

	// avoid timeouts like:
	// Error from server (InternalError): Internal error occurred: failed calling webhook "validate.nginx.ingress.kubernetes.io": Post "https://ingress-nginx-controller-admission.ingress-nginx.svc:443/networking/v1/ingresses?timeout=10s": dial tcp 10.107.218.58:443: i/o timeout
	// Error from server (InternalError): Internal error occurred: failed calling webhook "validate.nginx.ingress.kubernetes.io": Post "https://ingress-nginx-controller-admission.ingress-nginx.svc:443/networking/v1/ingresses?timeout=10s": context deadline exceeded
	if _, err = Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "wait", "--for=condition=ready", "--namespace=ingress-nginx", "pod", "--selector=app.kubernetes.io/component=controller", "--timeout=90s")); err != nil {
		t.Fatalf("failed waiting for ingress-nginx-controller : %v", err)
	}

	// use nginx ingress yaml that corresponds to k8s version
	// default: k8s >= v1.19, ingress api v1
	ingressYaml := "nginx-ingress-v1.yaml"
	ingressDNSYaml := "ingress-dns-example-v1.yaml"
	v, err := client.ServerVersion()
	if err != nil {
		t.Log("failed to get k8s version, assuming v1.19+ => ingress api v1")
	} else if semver.MustParseRange("<1.19.0")(semver.MustParse(fmt.Sprintf("%s.%s.0", v.Major, v.Minor))) {
		// legacy: k8s < v1.19 & ingress api v1beta1
		ingressYaml = "nginx-ingress-v1beta1.yaml"
		ingressDNSYaml = "ingress-dns-example-v1beta1.yaml"
	}

	// create networking.k8s.io/v1 ingress
	createv1Ingress := func() error {
		// apply networking.k8s.io/v1 ingress
		rr, err := Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "replace", "--force", "-f", filepath.Join(*testdataDir, ingressYaml)))
		if err != nil {
			return err
		}
		if rr.Stderr.String() != "" {
			t.Logf("%v: unexpected stderr: %s (may be temporary)", rr.Command(), rr.Stderr)
		}
		return nil
	}
	if err := retry.Expo(createv1Ingress, 1*time.Second, Seconds(90)); err != nil {
		t.Errorf("failed to create ingress: %v", err)
	}

	rr, err := Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "replace", "--force", "-f", filepath.Join(*testdataDir, "nginx-pod-svc.yaml")))
	if err != nil {
		t.Errorf("failed to kubectl replace nginx-pod-svc. args %q. %v", rr.Command(), err)
	}

	if _, err := PodWait(ctx, t, profile, "default", "run=nginx", Minutes(8)); err != nil {
		t.Fatalf("failed waiting for ngnix pod: %v", err)
	}
	if err := kapi.WaitForService(client, "default", "nginx", true, time.Millisecond*500, Minutes(10)); err != nil {
		t.Errorf("failed waiting for nginx service to be up: %v", err)
	}

	want := "Welcome to nginx!"
	addr := "http://127.0.0.1/"

	// check if the ingress can route nginx app with networking.k8s.io/v1 ingress
	checkv1Ingress := func() error {
		rr, err := Run(t, exec.CommandContext(ctx, Target(), "-p", profile, "ssh", fmt.Sprintf("curl -s %s -H 'Host: nginx.example.com'", addr)))
		if err != nil {
			return err
		}

		stderr := rr.Stderr.String()
		if rr.Stderr.String() != "" {
			t.Logf("debug: unexpected stderr for %v:\n%s", rr.Command(), stderr)
		}
		stdout := rr.Stdout.String()
		if !strings.Contains(stdout, want) {
			return fmt.Errorf("%v stdout = %q, want %q", rr.Command(), stdout, want)
		}
		return nil
	}
	if err := retry.Expo(checkv1Ingress, 500*time.Millisecond, Seconds(90)); err != nil {
		t.Errorf("failed to get expected response from %s within minikube: %v", addr, err)
	}

	if NeedsPortForward() {
		t.Skip("skipping ingress DNS test for any combination that needs port forwarding")
	}

	// check the ingress-dns addon here as well
	rr, err = Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "replace", "--force", "-f", filepath.Join(*testdataDir, ingressDNSYaml)))
	if err != nil {
		t.Errorf("failed to kubectl replace ingress-dns-example. args %q. %v", rr.Command(), err)
	}

	rr, err = Run(t, exec.CommandContext(ctx, Target(), "-p", profile, "ip"))
	if err != nil {
		t.Errorf("failed to retrieve minikube ip. args %q : %v", rr.Command(), err)
	}
	ip := strings.TrimSuffix(rr.Stdout.String(), "\n")

	rr, err = Run(t, exec.CommandContext(ctx, "nslookup", "hello-john.test", ip))
	if err != nil {
		t.Errorf("failed to nslookup hello-john.test host. args %q : %v", rr.Command(), err)
	}
	// nslookup should include info about the hello-john.test host, including minikube's ip
	if !strings.Contains(rr.Stdout.String(), ip) {
		t.Errorf("unexpected output from nslookup. stdout: %v\nstderr: %v", rr.Stdout.String(), rr.Stderr.String())
	}

	rr, err = Run(t, exec.CommandContext(ctx, Target(), "-p", profile, "addons", "disable", "ingress-dns", "--alsologtostderr", "-v=1"))
	if err != nil {
		t.Errorf("failed to disable ingress-dns addon. args %q : %v", rr.Command(), err)
	}

	rr, err = Run(t, exec.CommandContext(ctx, Target(), "-p", profile, "addons", "disable", "ingress", "--alsologtostderr", "-v=1"))
	if err != nil {
		t.Errorf("failed to disable ingress addon. args %q : %v", rr.Command(), err)
	}
}

// validateRegistryAddon tests the registry addon
func validateRegistryAddon(ctx context.Context, t *testing.T, profile string) {
	defer PostMortemLogs(t, profile)

	client, err := kapi.Client(profile)
	if err != nil {
		t.Fatalf("failed to get Kubernetes client for %s : %v", profile, err)
	}

	start := time.Now()
	if err := kapi.WaitForRCToStabilize(client, "kube-system", "registry", Minutes(6)); err != nil {
		t.Errorf("failed waiting for registry replicacontroller to stabilize: %v", err)
	}
	t.Logf("registry stabilized in %s", time.Since(start))

	if _, err := PodWait(ctx, t, profile, "kube-system", "actual-registry=true", Minutes(6)); err != nil {
		t.Fatalf("failed waiting for pod actual-registry: %v", err)
	}
	if _, err := PodWait(ctx, t, profile, "kube-system", "registry-proxy=true", Minutes(10)); err != nil {
		t.Fatalf("failed waiting for pod registry-proxy: %v", err)
	}

	// Test from inside the cluster (no curl available on busybox)
	rr, err := Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "delete", "po", "-l", "run=registry-test", "--now"))
	if err != nil {
		t.Logf("pre-cleanup %s failed: %v (not a problem)", rr.Command(), err)
	}

	rr, err = Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "run", "--rm", "registry-test", "--restart=Never", "--image=busybox", "-it", "--", "sh", "-c", "wget --spider -S http://registry.kube-system.svc.cluster.local"))
	if err != nil {
		t.Errorf("failed to hit registry.kube-system.svc.cluster.local. args %q failed: %v", rr.Command(), err)
	}
	want := "HTTP/1.1 200"
	if !strings.Contains(rr.Stdout.String(), want) {
		t.Errorf("expected curl response be %q, but got *%s*", want, rr.Stdout.String())
	}

	if NeedsPortForward() {
		t.Skipf("Unable to complete rest of the test due to connectivity assumptions")
	}

	// Test from outside the cluster
	rr, err = Run(t, exec.CommandContext(ctx, Target(), "-p", profile, "ip"))
	if err != nil {
		t.Fatalf("failed run minikube ip. args %q : %v", rr.Command(), err)
	}
	if rr.Stderr.String() != "" {
		t.Errorf("expected stderr to be -empty- but got: *%q* .  args %q", rr.Stderr, rr.Command())
	}

	endpoint := fmt.Sprintf("http://%s:%d", strings.TrimSpace(rr.Stdout.String()), 5000)
	u, err := url.Parse(endpoint)
	if err != nil {
		t.Fatalf("failed to parse %q: %v", endpoint, err)
	}

	checkExternalAccess := func() error {
		resp, err := retryablehttp.Get(u.String())
		if err != nil {
			return err
		}
		if resp.StatusCode != http.StatusOK {
			return fmt.Errorf("%s = status code %d, want %d", u, resp.StatusCode, http.StatusOK)
		}
		return nil
	}

	if err := retry.Expo(checkExternalAccess, 500*time.Millisecond, Seconds(150)); err != nil {
		t.Errorf("failed to check external access to %s: %v", u.String(), err.Error())
	}

	rr, err = Run(t, exec.CommandContext(ctx, Target(), "-p", profile, "addons", "disable", "registry", "--alsologtostderr", "-v=1"))
	if err != nil {
		t.Errorf("failed to disable registry addon. args %q: %v", rr.Command(), err)
	}
}

// validateMetricsServerAddon tests the metrics server addon by making sure "kubectl top pods" returns a sensible result
func validateMetricsServerAddon(ctx context.Context, t *testing.T, profile string) {
	defer PostMortemLogs(t, profile)

	client, err := kapi.Client(profile)
	if err != nil {
		t.Fatalf("failed to get Kubernetes client for %s: %v", profile, err)
	}

	start := time.Now()
	if err := kapi.WaitForDeploymentToStabilize(client, "kube-system", "metrics-server", Minutes(6)); err != nil {
		t.Errorf("failed waiting for metrics-server deployment to stabilize: %v", err)
	}
	t.Logf("metrics-server stabilized in %s", time.Since(start))

	if _, err := PodWait(ctx, t, profile, "kube-system", "k8s-app=metrics-server", Minutes(6)); err != nil {
		t.Fatalf("failed waiting for k8s-app=metrics-server pod: %v", err)
	}

	want := "CPU(cores)"
	checkMetricsServer := func() error {
		rr, err := Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "top", "pods", "-n", "kube-system"))
		if err != nil {
			return err
		}
		if rr.Stderr.String() != "" {
			t.Logf("%v: unexpected stderr: %s", rr.Command(), rr.Stderr)
		}
		if !strings.Contains(rr.Stdout.String(), want) {
			return fmt.Errorf("%v stdout = %q, want %q", rr.Command(), rr.Stdout, want)
		}
		return nil
	}

	if err := retry.Expo(checkMetricsServer, time.Second*3, Minutes(6)); err != nil {
		t.Errorf("failed checking metric server: %v", err.Error())
	}

	rr, err := Run(t, exec.CommandContext(ctx, Target(), "-p", profile, "addons", "disable", "metrics-server", "--alsologtostderr", "-v=1"))
	if err != nil {
		t.Errorf("failed to disable metrics-server addon: args %q: %v", rr.Command(), err)
	}
}

// validateHelmTillerAddon tests the helm tiller addon by running "helm version" inside the cluster
func validateHelmTillerAddon(ctx context.Context, t *testing.T, profile string) {

	defer PostMortemLogs(t, profile)

	if arm64Platform() {
		t.Skip("skip Helm test on arm64")
	}

	client, err := kapi.Client(profile)
	if err != nil {
		t.Fatalf("failed to get Kubernetes client for %s: %v", profile, err)
	}

	start := time.Now()
	if err := kapi.WaitForDeploymentToStabilize(client, "kube-system", "tiller-deploy", Minutes(6)); err != nil {
		t.Errorf("failed waiting for tiller-deploy deployment to stabilize: %v", err)
	}
	t.Logf("tiller-deploy stabilized in %s", time.Since(start))

	if _, err := PodWait(ctx, t, profile, "kube-system", "app=helm", Minutes(6)); err != nil {
		t.Fatalf("failed waiting for helm pod: %v", err)
	}

	if NoneDriver() {
		_, err := exec.LookPath("socat")
		if err != nil {
			t.Skipf("socat is required by kubectl to complete this test")
		}
	}

	want := "Server: &version.Version"
	// Test from inside the cluster (`helm version` use pod.list permission.)
	checkHelmTiller := func() error {

		rr, err := Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "run", "--rm", "helm-test", "--restart=Never", "--image=alpine/helm:2.16.3", "-it", "--namespace=kube-system", "--", "version"))
		if err != nil {
			return err
		}
		if rr.Stderr.String() != "" {
			t.Logf("%v: unexpected stderr: %s", rr.Command(), rr.Stderr)
		}
		if !strings.Contains(rr.Stdout.String(), want) {
			return fmt.Errorf("%v stdout = %q, want %q", rr.Command(), rr.Stdout, want)
		}
		return nil
	}

	if err := retry.Expo(checkHelmTiller, 500*time.Millisecond, Minutes(2)); err != nil {
		t.Errorf("failed checking helm tiller: %v", err.Error())
	}

	rr, err := Run(t, exec.CommandContext(ctx, Target(), "-p", profile, "addons", "disable", "helm-tiller", "--alsologtostderr", "-v=1"))
	if err != nil {
		t.Errorf("failed disabling helm-tiller addon. arg %q.s %v", rr.Command(), err)
	}
}

// validateOlmAddon tests the OLM addon
func validateOlmAddon(ctx context.Context, t *testing.T, profile string) {
	defer PostMortemLogs(t, profile)

	start := time.Now()

	if _, err := Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "wait", "--for=condition=ready", "--namespace=olm", "pod", "--selector=app=catalog-operator", "--timeout=90s")); err != nil {
		t.Fatalf("failed waititing for pod catalog-operator: %v", err)
	}
	t.Logf("catalog-operator stabilized in %s", time.Since(start))

	if _, err := Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "wait", "--for=condition=ready", "--namespace=olm", "pod", "--selector=app=olm-operator", "--timeout=90s")); err != nil {
		t.Fatalf("failed waititing for pod olm-operator: %v", err)
	}
	t.Logf("olm-operator stabilized in %s", time.Since(start))

	if _, err := Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "wait", "--for=condition=ready", "--namespace=olm", "pod", "--selector=app=packageserver", "--timeout=90s")); err != nil {
		t.Fatalf("failed waititing for pod olm-operator: %v", err)
	}
	t.Logf("packageserver stabilized in %s", time.Since(start))

	if _, err := Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "wait", "--for=condition=ready", "--namespace=olm", "pod", "--selector=olm.catalogSource=operatorhubio-catalog", "--timeout=90s")); err != nil {
		t.Fatalf("failed waititing for pod operatorhubio-catalog: %v", err)
	}
	t.Logf("operatorhubio-catalog stabilized in %s", time.Since(start))

	// Install one sample Operator such as etcd
	rr, err := Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "create", "-f", filepath.Join(*testdataDir, "etcd.yaml")))
	if err != nil {
		t.Logf("etcd operator installation with %s failed: %v", rr.Command(), err)
	}

	want := "Succeeded"
	checkOperatorInstalled := func() error {
		rr, err := Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "get", "csv", "-n", "my-etcd"))
		if err != nil {
			return err
		}
		if rr.Stderr.String() != "" {
			t.Logf("%v: unexpected stderr: %s", rr.Command(), rr.Stderr)
		}
		if !strings.Contains(rr.Stdout.String(), want) {
			return fmt.Errorf("%v stdout = %q, want %q", rr.Command(), rr.Stdout, want)
		}
		return nil
	}
	// Operator installation takes a while
	if err := retry.Expo(checkOperatorInstalled, time.Second*3, Minutes(10)); err != nil {
		t.Errorf("failed checking operator installed: %v", err.Error())
	}
}

// validateCSIDriverAndSnapshots tests the csi hostpath driver by creating a persistent volume, snapshotting it and restoring it.
func validateCSIDriverAndSnapshots(ctx context.Context, t *testing.T, profile string) {
	defer PostMortemLogs(t, profile)

	client, err := kapi.Client(profile)
	if err != nil {
		t.Fatalf("failed to get Kubernetes client for %s: %v", profile, err)
	}

	start := time.Now()
	if err := kapi.WaitForPods(client, "kube-system", "kubernetes.io/minikube-addons=csi-hostpath-driver", Minutes(6)); err != nil {
		t.Errorf("failed waiting for csi-hostpath-driver pods to stabilize: %v", err)
	}
	t.Logf("csi-hostpath-driver pods stabilized in %s", time.Since(start))

	// create sample PVC
	rr, err := Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "create", "-f", filepath.Join(*testdataDir, "csi-hostpath-driver", "pvc.yaml")))
	if err != nil {
		t.Logf("creating sample PVC with %s failed: %v", rr.Command(), err)
	}

	if err := PVCWait(ctx, t, profile, "default", "hpvc", Minutes(6)); err != nil {
		t.Fatalf("failed waiting for PVC hpvc: %v", err)
	}

	// create sample pod with the PVC
	rr, err = Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "create", "-f", filepath.Join(*testdataDir, "csi-hostpath-driver", "pv-pod.yaml")))
	if err != nil {
		t.Logf("creating pod with %s failed: %v", rr.Command(), err)
	}

	if _, err := PodWait(ctx, t, profile, "default", "app=task-pv-pod", Minutes(6)); err != nil {
		t.Fatalf("failed waiting for pod task-pv-pod: %v", err)
	}

	// create volume snapshot
	rr, err = Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "create", "-f", filepath.Join(*testdataDir, "csi-hostpath-driver", "snapshot.yaml")))
	if err != nil {
		t.Logf("creating pod with %s failed: %v", rr.Command(), err)
	}

	if err := VolumeSnapshotWait(ctx, t, profile, "default", "new-snapshot-demo", Minutes(6)); err != nil {
		t.Fatalf("failed waiting for volume snapshot new-snapshot-demo: %v", err)
	}

	// delete pod
	rr, err = Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "delete", "pod", "task-pv-pod"))
	if err != nil {
		t.Logf("deleting pod with %s failed: %v", rr.Command(), err)
	}

	// delete pvc
	rr, err = Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "delete", "pvc", "hpvc"))
	if err != nil {
		t.Logf("deleting pod with %s failed: %v", rr.Command(), err)
	}

	// restore pv from snapshot
	rr, err = Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "create", "-f", filepath.Join(*testdataDir, "csi-hostpath-driver", "pvc-restore.yaml")))
	if err != nil {
		t.Logf("creating pvc with %s failed: %v", rr.Command(), err)
	}

	if err = PVCWait(ctx, t, profile, "default", "hpvc-restore", Minutes(6)); err != nil {
		t.Fatalf("failed waiting for PVC hpvc-restore: %v", err)
	}

	// create pod from restored snapshot
	rr, err = Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "create", "-f", filepath.Join(*testdataDir, "csi-hostpath-driver", "pv-pod-restore.yaml")))
	if err != nil {
		t.Logf("creating pod with %s failed: %v", rr.Command(), err)
	}

	if _, err := PodWait(ctx, t, profile, "default", "app=task-pv-pod-restore", Minutes(6)); err != nil {
		t.Fatalf("failed waiting for pod task-pv-pod-restore: %v", err)
	}

	// CLEANUP
	rr, err = Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "delete", "pod", "task-pv-pod-restore"))
	if err != nil {
		t.Logf("cleanup with %s failed: %v", rr.Command(), err)
	}
	rr, err = Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "delete", "pvc", "hpvc-restore"))
	if err != nil {
		t.Logf("cleanup with %s failed: %v", rr.Command(), err)
	}
	rr, err = Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "delete", "volumesnapshot", "new-snapshot-demo"))
	if err != nil {
		t.Logf("cleanup with %s failed: %v", rr.Command(), err)
	}
	rr, err = Run(t, exec.CommandContext(ctx, Target(), "-p", profile, "addons", "disable", "csi-hostpath-driver", "--alsologtostderr", "-v=1"))
	if err != nil {
		t.Errorf("failed to disable csi-hostpath-driver addon: args %q: %v", rr.Command(), err)
	}
	rr, err = Run(t, exec.CommandContext(ctx, Target(), "-p", profile, "addons", "disable", "volumesnapshots", "--alsologtostderr", "-v=1"))
	if err != nil {
		t.Errorf("failed to disable volumesnapshots addon: args %q: %v", rr.Command(), err)
	}
}

// validateGCPAuthAddon tests the GCP Auth addon with either phony or real credentials and makes sure the files are mounted into pods correctly
func validateGCPAuthAddon(ctx context.Context, t *testing.T, profile string) {
	defer PostMortemLogs(t, profile)

	// schedule a pod to check environment variables
	rr, err := Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "create", "-f", filepath.Join(*testdataDir, "busybox.yaml")))
	if err != nil {
		t.Fatalf("%s failed: %v", rr.Command(), err)
	}

	// 8 minutes, because 4 is not enough for images to pull in all cases.
	names, err := PodWait(ctx, t, profile, "default", "integration-test=busybox", Minutes(8))
	if err != nil {
		t.Fatalf("wait: %v", err)
	}

	// Use this pod to confirm that the env vars are set correctly
	rr, err = Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "exec", names[0], "--", "/bin/sh", "-c", "printenv GOOGLE_APPLICATION_CREDENTIALS"))
	if err != nil {
		t.Fatalf("printenv creds: %v", err)
	}

	got := strings.TrimSpace(rr.Stdout.String())
	expected := "/google-app-creds.json"
	if got != expected {
		t.Errorf("'printenv GOOGLE_APPLICATION_CREDENTIALS' returned %s, expected %s", got, expected)
	}

	if !detect.IsOnGCE() || detect.IsCloudShell() {
		// Make sure the file contents are correct
		rr, err = Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "exec", names[0], "--", "/bin/sh", "-c", "cat /google-app-creds.json"))
		if err != nil {
			t.Fatalf("cat creds: %v", err)
		}

		var gotJSON map[string]string
		err = json.Unmarshal(bytes.TrimSpace(rr.Stdout.Bytes()), &gotJSON)
		if err != nil {
			t.Fatalf("unmarshal json: %v", err)
		}
		expectedJSON := map[string]string{
			"client_id":        "haha",
			"client_secret":    "nice_try",
			"quota_project_id": "this_is_fake",
			"refresh_token":    "maybe_next_time",
			"type":             "authorized_user",
		}

		if !reflect.DeepEqual(gotJSON, expectedJSON) {
			t.Fatalf("unexpected creds file: got %v, expected %v", gotJSON, expectedJSON)
		}
	}

	// Check the GOOGLE_CLOUD_PROJECT env var as well
	rr, err = Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "exec", names[0], "--", "/bin/sh", "-c", "printenv GOOGLE_CLOUD_PROJECT"))
	if err != nil {
		t.Fatalf("print env project: %v", err)
	}

	got = strings.TrimSpace(rr.Stdout.String())
	expected = "this_is_fake"

	if got != expected {
		t.Errorf("'printenv GOOGLE_CLOUD_PROJECT' returned %s, expected %s", got, expected)
	}

	disableGCPAuth := func() error {
		_, err = Run(t, exec.CommandContext(ctx, Target(), "-p", profile, "addons", "disable", "gcp-auth", "--alsologtostderr", "-v=1"))
		if err != nil {
			return err
		}
		return nil
	}

	if err := retry.Expo(disableGCPAuth, Minutes(2), Minutes(10), 5); err != nil {
		t.Errorf("failed to disable GCP auth addon: %v", err)
	}

	// If we're on GCE, we have proper credentials and can test the registry secrets with an artifact registry image
	if detect.IsOnGCE() && !detect.IsCloudShell() {
		os.Unsetenv("GOOGLE_APPLICATION_CREDENTIALS")
		os.Unsetenv("GOOGLE_CLOUD_PROJECT")
		args := []string{"-p", profile, "addons", "enable", "gcp-auth"}
		rr, err := Run(t, exec.CommandContext(ctx, Target(), args...))
		if err != nil {
			t.Errorf("%s failed: %v", rr.Command(), err)
		} else if !strings.Contains(rr.Output(), "It seems that you are running in GCE") {
			t.Errorf("Unexpected error message: %v", rr.Output())
		}
		_, err = Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "apply", "-f", filepath.Join(*testdataDir, "private-image.yaml")))
		if err != nil {
			t.Fatalf("print env project: %v", err)
		}

		// Make sure the pod is up and running, which means we successfully pulled the private image down
		// 8 minutes, because 4 is not enough for images to pull in all cases.
		_, err = PodWait(ctx, t, profile, "default", "integration-test=private-image", Minutes(8))
		if err != nil {
			t.Fatalf("wait for private image: %v", err)
		}

		// Try it with a European mirror as well
		_, err = Run(t, exec.CommandContext(ctx, "kubectl", "--context", profile, "apply", "-f", filepath.Join(*testdataDir, "private-image-eu.yaml")))
		if err != nil {
			t.Fatalf("print env project: %v", err)
		}

		_, err = PodWait(ctx, t, profile, "default", "integration-test=private-image-eu", Minutes(8))
		if err != nil {
			t.Fatalf("wait for private image: %v", err)
		}
	}
}


text:
#region S# License

#endregion S# License
namespace StockSharp.Algo
{
	using System;
	using System.Collections.Generic;
	using System.Linq;
	using System.Security;

	using Ecng.Common;
	using Ecng.ComponentModel;
	using Ecng.Serialization;
	using Ecng.Collections;

	using StockSharp.Algo.Candles;
	using StockSharp.Algo.Candles.Compression;
	using StockSharp.Algo.Commissions;
	using StockSharp.Algo.Latency;
	using StockSharp.Algo.PnL;
	using StockSharp.Algo.Risk;
	using StockSharp.Algo.Slippage;
	using StockSharp.Algo.Storages;
	using StockSharp.BusinessEntities;
	using StockSharp.Logging;
	using StockSharp.Messages;
	using StockSharp.Localization;

	/// <summary>
	/// The class to create connections to trading systems.
	/// </summary>
	public partial class Connector : BaseLogReceiver, IConnector, ICandleManager, IMarketDataProvider, ISubscriptionProvider
	{
		private readonly EntityCache _entityCache;
		private readonly SubscriptionManager _subscriptionManager;

		// backward compatibility for NewXXX events
		private readonly CachedSynchronizedSet<Security> _existingSecurities = new();
		private readonly CachedSynchronizedSet<Portfolio> _existingPortfolios = new();
		private readonly CachedSynchronizedSet<Position> _existingPositions = new();

		private bool _isDisposing;

		/// <summary>
		/// Initializes a new instance of the <see cref="Connector"/>.
		/// </summary>
		public Connector()
			: this(new InMemorySecurityStorage(), new InMemoryPositionStorage(), new InMemoryExchangeInfoProvider())
		{
		}

		/// <summary>
		/// Initializes a new instance of the <see cref="Connector"/>.
		/// </summary>
		/// <param name="securityStorage">Securities meta info storage.</param>
		/// <param name="positionStorage">Position storage.</param>
		/// <param name="exchangeInfoProvider">Exchanges and trading boards provider.</param>
		/// <param name="storageRegistry">The storage of market data.</param>
		/// <param name="snapshotRegistry">Snapshot storage registry.</param>
		/// <param name="buffer">Storage buffer.</param>
		/// <param name="initAdapter">Initialize basket adapter.</param>
		/// <param name="initChannels">Initialize channels.</param>
		public Connector(ISecurityStorage securityStorage, IPositionStorage positionStorage,
			IExchangeInfoProvider exchangeInfoProvider, IStorageRegistry storageRegistry = null,
			SnapshotRegistry snapshotRegistry = null, StorageBuffer buffer = null, bool initAdapter = true, bool initChannels = true)
		{
			Buffer = buffer;

			SecurityStorage = securityStorage ?? throw new ArgumentNullException(nameof(securityStorage));
			PositionStorage = positionStorage ?? throw new ArgumentNullException(nameof(positionStorage));

			_entityCache = new EntityCache(this, TryGetSecurity, new EntityFactory(), exchangeInfoProvider, PositionStorage);

			_subscriptionManager = new SubscriptionManager(this);

			//SupportLevel1DepthBuilder = true;
			SupportFilteredMarketDepth = true;

			if (initChannels)
			{
				InMessageChannel = new InMemoryMessageChannel(new MessageByOrderQueue(), $"Connector In ({Name})", RaiseError);
				OutMessageChannel = new InMemoryMessageChannel(new MessageByOrderQueue(), $"Connector Out ({Name})", RaiseError);
			}

			SnapshotRegistry = snapshotRegistry;

			if (initAdapter)
			{
				Adapter = new BasketMessageAdapter(new MillisecondIncrementalIdGenerator(), new CandleBuilderProvider(ExchangeInfoProvider), new InMemorySecurityMessageAdapterProvider(), new InMemoryPortfolioMessageAdapterProvider())
				{
					StorageSettings = { StorageRegistry = storageRegistry }
				};
			}
		}

		/// <summary>
		/// Initializes a new instance of the <see cref="Connector"/>.
		/// </summary>
		/// <param name="entityRegistry">The storage of trade objects.</param>
		/// <param name="storageRegistry">The storage of market data.</param>
		/// <param name="snapshotRegistry">Snapshot storage registry.</param>
		/// <param name="buffer">Storage buffer.</param>
		[Obsolete]
		public Connector(IEntityRegistry entityRegistry, IStorageRegistry storageRegistry, SnapshotRegistry snapshotRegistry, StorageBuffer buffer = null)
			: this(entityRegistry.Securities, entityRegistry.PositionStorage, storageRegistry.CheckOnNull().ExchangeInfoProvider, storageRegistry, snapshotRegistry, buffer)
		{
		}

		/// <summary>
		/// Securities meta info storage.
		/// </summary>
		public ISecurityStorage SecurityStorage { get; }

		/// <summary>
		/// Position storage.
		/// </summary>
		public IPositionStorage PositionStorage { get; }

		/// <summary>
		/// Exchanges and trading boards provider.
		/// </summary>
		public IExchangeInfoProvider ExchangeInfoProvider => _entityCache.ExchangeInfoProvider;

		/// <summary>
		/// The storage of market data.
		/// </summary>
		public IStorageRegistry StorageRegistry => Adapter?.StorageSettings.StorageRegistry;

		/// <summary>
		/// Snapshot storage registry.
		/// </summary>
		public SnapshotRegistry SnapshotRegistry { get; private set; }

		private IBasketSecurityProcessorProvider _basketSecurityProcessorProvider = new BasketSecurityProcessorProvider();

		/// <summary>
		/// Basket security processors provider.
		/// </summary>
		public IBasketSecurityProcessorProvider BasketSecurityProcessorProvider
		{
			get => _basketSecurityProcessorProvider;
			set => _basketSecurityProcessorProvider = value ?? throw new ArgumentNullException(nameof(value));
		}

		/// <summary>
		/// Restore subscription on reconnect.
		/// </summary>
		/// <remarks>
		/// Normal case connect/disconnect.
		/// </remarks>
		public bool IsRestoreSubscriptionOnNormalReconnect { get; set; } = true;

		/// <summary>
		/// Send unsubscribe on disconnect command.
		/// </summary>
		/// <remarks>By default is <see langword="true"/>.</remarks>
		public bool IsAutoUnSubscribeOnDisconnect { get; set; } = true;

		/// <summary>
		/// Subscribe for new portfolios.
		/// </summary>
		/// <remarks>By default is <see langword="true"/>.</remarks>
		public bool IsAutoPortfoliosSubscribe { get; set; } = true;

		/// <summary>
		/// Settings of the connection control <see cref="IConnector"/> to the trading system.
		/// </summary>
		[Obsolete("Use exact IMessageAdapter to set reconnecting settings.")]
		public ReConnectionSettings ReConnectionSettings { get; } = new ReConnectionSettings();

		/// <summary>
		/// Entity factory (<see cref="Security"/>, <see cref="Order"/> etc.).
		/// </summary>
		public IEntityFactory EntityFactory => _entityCache.EntityFactory;

		/// <summary>
		/// Number of tick trades for storage. The default is 100000. If the value is set to <see cref="int.MaxValue"/>, the trades will not be deleted. If the value is set to 0, then the trades will not be stored.
		/// </summary>
		public int TradesKeepCount
		{
			get => _entityCache.TradesKeepCount;
			set => _entityCache.TradesKeepCount = value;
		}

		/// <summary>
		/// The number of orders for storage. The default is 1000. If the value is set to <see cref="int.MaxValue"/>, then the orders will not be deleted. If the value is set to 0, then the orders will not be stored.
		/// </summary>
		public int OrdersKeepCount
		{
			get => _entityCache.OrdersKeepCount;
			set => _entityCache.OrdersKeepCount = value;
		}

		/// <inheritdoc />
		public IdGenerator TransactionIdGenerator
		{
			get => Adapter.TransactionIdGenerator;
			set => Adapter.TransactionIdGenerator = value;
		}

		private SecurityIdGenerator _securityIdGenerator = new();

		/// <summary>
		/// The instrument identifiers generator <see cref="Security.Id"/>.
		/// </summary>
		public SecurityIdGenerator SecurityIdGenerator
		{
			get => _securityIdGenerator;
			set => _securityIdGenerator = value ?? throw new ArgumentNullException(nameof(value));
		}

		private bool _overrideSecurityData;

		/// <summary>
		/// Override previous security data by new values.
		/// </summary>
		public bool OverrideSecurityData
		{
			get => _overrideSecurityData;
			set
			{
				_overrideSecurityData = value;

				if (StorageAdapter != null)
					StorageAdapter.OverrideSecurityData = value;
			}
		}

		/// <inheritdoc />
		public IEnumerable<ExchangeBoard> ExchangeBoards => ExchangeInfoProvider.Boards;

		/// <inheritdoc />
		public IEnumerable<Security> Securities => _existingSecurities.Cache;

		int ISecurityProvider.Count => SecurityStorage.Count;

		private Action<IEnumerable<Security>> _added;

		event Action<IEnumerable<Security>> ISecurityProvider.Added
		{
			add => _added += value;
			remove => _added -= value;
		}

		private Action<IEnumerable<Security>> _removed;

		event Action<IEnumerable<Security>> ISecurityProvider.Removed
		{
			add => _removed += value;
			remove => _removed -= value;
		}

		private Action _cleared;

		event Action ISecurityProvider.Cleared
		{
			add => _cleared += value;
			remove => _cleared -= value;
		}

		/// <inheritdoc />
		public Security LookupById(SecurityId id) => SecurityStorage.LookupById(id);

		IEnumerable<Security> ISecurityProvider.Lookup(SecurityLookupMessage criteria) => SecurityStorage.Lookup(criteria);

		SecurityMessage ISecurityMessageProvider.LookupMessageById(SecurityId id)
			=> SecurityStorage.LookupMessageById(id);

		IEnumerable<SecurityMessage> ISecurityMessageProvider.LookupMessages(SecurityLookupMessage criteria)
			=> SecurityStorage.LookupMessages(criteria);

		/// <inheritdoc />
		public SessionStates? GetSessionState(ExchangeBoard board) => _entityCache.GetSessionState(board);

		/// <inheritdoc />
		[Obsolete("Use NewOrder event to collect data.")]
		public IEnumerable<Order> Orders => _entityCache.Orders;

		/// <inheritdoc />
		[Obsolete("Use NewStopOrder event to collect data.")]
		public IEnumerable<Order> StopOrders => Orders.Where(o => o.Type == OrderTypes.Conditional);

		/// <inheritdoc />
		[Obsolete("Use OrderRegisterFailed event to collect data.")]
		public IEnumerable<OrderFail> OrderRegisterFails => _entityCache.OrderRegisterFails;

		/// <inheritdoc />
		[Obsolete("Use OrderCancelFailed event to collect data.")]
		public IEnumerable<OrderFail> OrderCancelFails => _entityCache.OrderCancelFails;

		/// <inheritdoc />
		[Obsolete("Use NewTrade event to collect data.")]
		public IEnumerable<Trade> Trades => _entityCache.Trades;

		/// <inheritdoc />
		[Obsolete("Use NewMyTrade event to collect data.")]
		public IEnumerable<MyTrade> MyTrades => _entityCache.MyTrades;

		/// <inheritdoc />
		[Obsolete("Use NewNews event to collect data.")]
		public IEnumerable<News> News => _entityCache.News;

		/// <inheritdoc />
		public IEnumerable<Portfolio> Portfolios => _existingPortfolios.Cache;

		/// <inheritdoc />
		public IEnumerable<Position> Positions => _existingPositions.Cache;

		/// <summary>
		/// Risk control manager.
		/// </summary>
		public virtual IRiskManager RiskManager { get; set; } = new RiskManager();

		/// <summary>
		/// Orders registration delay calculation manager.
		/// </summary>
		public ILatencyManager LatencyManager
		{
			get => Adapter.LatencyManager;
			set => Adapter.LatencyManager = value;
		}

		/// <summary>
		/// The profit-loss manager.
		/// </summary>
		public IPnLManager PnLManager
		{
			get => Adapter.PnLManager;
			set => Adapter.PnLManager = value;
		}

		/// <summary>
		/// The commission calculating manager.
		/// </summary>
		public ICommissionManager CommissionManager
		{
			get => Adapter.CommissionManager;
			set => Adapter.CommissionManager = value;
		}

		/// <summary>
		/// Slippage manager.
		/// </summary>
		public ISlippageManager SlippageManager
		{
			get => Adapter.SlippageManager;
			set => Adapter.SlippageManager = value;
		}

		private ConnectionStates _connectionState;

		/// <inheritdoc />
		public ConnectionStates ConnectionState
		{
			get => _connectionState;
			private set
			{
				_connectionState = value;
				_stateChanged?.Invoke();
			}
		}

		/// <summary>
		/// Use orders log to create market depths. Disabled by default.
		/// </summary>
		[Obsolete("Use MarketDataMessage.BuildFrom=OrderLog instead.")]
		public bool CreateDepthFromOrdersLog { get; set; }

		/// <summary>
		/// Use orders log to create ticks. Disabled by default.
		/// </summary>
		[Obsolete("Use MarketDataMessage.BuildFrom=OrderLog instead.")]
		public bool CreateTradesFromOrdersLog { get; set; }

		/// <summary>
		/// To update <see cref="Security.LastTrade"/>, <see cref="Security.BestBid"/>, <see cref="Security.BestAsk"/> at each update of order book and/or trades. By default is enabled.
		/// </summary>
		public bool UpdateSecurityLastQuotes { get; set; } = true;

		/// <summary>
		/// To update <see cref="Security"/> fields when the <see cref="Level1ChangeMessage"/> message appears. By default is enabled.
		/// </summary>
		public bool UpdateSecurityByLevel1 { get; set; } = true;

		/// <summary>
		/// To update <see cref="Security"/> fields when the <see cref="SecurityMessage"/> message appears. By default is enabled.
		/// </summary>
		public bool UpdateSecurityByDefinition { get; set; } = true;

		/// <summary>
		/// To update the order book for the instrument when the <see cref="Level1ChangeMessage"/> message appears. By default is enabled.
		/// </summary>
		[DisplayNameLoc(LocalizedStrings.Str200Key)]
		[DescriptionLoc(LocalizedStrings.Str201Key)]
		[Obsolete("Use SupportLevel1DepthBuilder property.")]
		public bool CreateDepthFromLevel1
		{
			get => SupportLevel1DepthBuilder;
			set => SupportLevel1DepthBuilder = value;
		}

		/// <summary>
		/// Create a combined security for securities from different boards.
		/// </summary>
		[DisplayNameLoc(LocalizedStrings.Str197Key)]
		[DescriptionLoc(LocalizedStrings.Str198Key)]
		[Obsolete("Use SupportAssociatedSecurity property.")]
		public bool CreateAssociatedSecurity
		{
			get => SupportAssociatedSecurity;
			set => SupportAssociatedSecurity = value;
		}

		/// <summary>
		/// The number of errors passed through the <see cref="Connector.Error"/> event.
		/// </summary>
		public int ErrorCount { get; private set; }

		private TimeSpan _marketTimeChangedInterval = TimeSpan.FromMilliseconds(10);

		/// <summary>
		/// The <see cref="TimeMessage"/> message generating Interval. The default is 10 milliseconds.
		/// </summary>
		[CategoryLoc(LocalizedStrings.Str186Key)]
		[DisplayNameLoc(LocalizedStrings.TimeIntervalKey)]
		[DescriptionLoc(LocalizedStrings.Str195Key)]
		public virtual TimeSpan MarketTimeChangedInterval
		{
			get => _marketTimeChangedInterval;
			set
			{
				if (value <= TimeSpan.Zero)
					throw new ArgumentOutOfRangeException(nameof(value), value, LocalizedStrings.Str196);

				_marketTimeChangedInterval = value;
			}
		}

		/// <summary>
		/// Increment periodically <see cref="MarketTimeChangedInterval"/> value of <see cref="ILogSource.CurrentTime"/>.
		/// </summary>
		public bool TimeChange { get; set; } = true;

		/// <summary>
		/// Process strategies positions and store it into <see cref="Positions"/>.
		/// </summary>
		public bool KeepStrategiesPositions { get; set; }

		private readonly CachedSynchronizedSet<MessageTypes> _lookupMessagesOnConnect = new(new[]
		{
			MessageTypes.SecurityLookup,
			MessageTypes.PortfolioLookup,
			MessageTypes.OrderStatus,
			MessageTypes.TimeFrameLookup,
		});

		/// <summary>
		/// Send lookup messages on connect. By default is <see langword="true"/>.
		/// </summary>
		public ISet<MessageTypes> LookupMessagesOnConnect => _lookupMessagesOnConnect;

		/// <inheritdoc />
		public void Connect()
		{
			this.AddInfoLog(nameof(Connect));

			try
			{
				if (ConnectionState is not ConnectionStates.Disconnected and not ConnectionStates.Failed)
				{
					this.AddWarningLog(LocalizedStrings.Str1095Params, ConnectionState);
					return;
				}

				ConnectionState = ConnectionStates.Connecting;

				OnConnect();
			}
			catch (Exception ex)
			{
				RaiseConnectionError(ex);
			}
		}

		/// <summary>
		/// Connect to trading system.
		/// </summary>
		protected virtual void OnConnect()
		{
			if (TimeChange)
				CreateTimer();

			if (!IsRestoreSubscriptionOnNormalReconnect)
				_subscriptionManager.ClearCache();

			SendInMessage(new ConnectMessage());
		}

		/// <inheritdoc />
		public void Disconnect()
		{
			this.AddInfoLog(nameof(Disconnect));

			if (ConnectionState != ConnectionStates.Connected)
			{
				this.AddWarningLog(LocalizedStrings.Str1096Params, ConnectionState);
				return;
			}

			ConnectionState = ConnectionStates.Disconnecting;

			try
			{
				OnDisconnect();
			}
			catch (Exception ex)
			{
				RaiseConnectionError(ex);
			}
		}

		/// <summary>
		/// Disconnect from trading system.
		/// </summary>
		protected virtual void OnDisconnect()
		{
			if (IsAutoUnSubscribeOnDisconnect)
				_subscriptionManager.UnSubscribeAll();

			SendInMessage(new DisconnectMessage());
		}

		/// <inheritdoc />
		public Position GetPosition(Portfolio portfolio, Security security, string strategyId = "", Sides? side = null, string clientCode = "", string depoName = "", TPlusLimits? limitType = null)
		{
			return GetPosition(portfolio, security, strategyId, side, clientCode, depoName, limitType, string.Empty);
		}

		private Position GetPosition(Portfolio portfolio, Security security, string strategyId, Sides? side, string clientCode, string depoName, TPlusLimits? limitType, string description)
		{
			if (portfolio == null)
				throw new ArgumentNullException(nameof(portfolio));

			if (security == null)
				throw new ArgumentNullException(nameof(security));

			var position = PositionStorage.GetOrCreatePosition(portfolio, security, strategyId, side, clientCode, depoName, limitType, (pf, sec, sid, sd, clCode, ddep, limit) =>
			{
				var p = EntityFactory.CreatePosition(portfolio, security);

				p.DepoName = depoName;
				p.LimitType = limitType;
				p.Description = description;
				p.ClientCode = clientCode;
				p.StrategyId = strategyId;
				p.Side = side;

				return p;
			}, out _);

			if (_existingPositions.TryAdd(position))
				RaiseNewPosition(position);

			return position;
		}

		private MarketDepth GetMarketDepth(Security security, QuoteChangeMessage message)
		{
			var depth = _entityCache.GetMarketDepth(security, message, out var isNew);

			if (isNew)
			{
				if (message.IsFiltered)
					RaiseFilteredMarketDepthChanged(depth);
				else
					RaiseNewMarketDepth(depth);
			}

			return depth;
		}

		[Obsolete]
		private MarketDepth GetMarketDepth(Security security, bool isFiltered)
		{
			return GetMarketDepth(security, new QuoteChangeMessage
			{
				IsFiltered = isFiltered,
				SecurityId = security.ToSecurityId(),
				ServerTime = CurrentTime,
				LocalTime = CurrentTime,
			});
		}

		/// <inheritdoc />
		[Obsolete("Use MarketDepthReceived event.")]
		public MarketDepth GetMarketDepth(Security security) => GetMarketDepth(security, false);

		/// <inheritdoc />
		[Obsolete("Use MarketDepthReceived event.")]
		public MarketDepth GetFilteredMarketDepth(Security security) => GetMarketDepth(security, true);

		/// <summary>
		/// Check <see cref="Order.Price"/> and <see cref="Order.Volume"/> are they multiply to step.
		/// </summary>
		public bool CheckSteps { get; set; }

		/// <inheritdoc />
		public void RegisterOrder(Order order)
		{
			try
			{
				this.AddOrderInfoLog(order, nameof(RegisterOrder));

				CheckOnNew(order);

				if (order.Type != OrderTypes.Conditional)
				{
					if (order.Volume == 0)
						throw new ArgumentException(LocalizedStrings.Str894, nameof(order));

					if (order.Volume < 0)
						throw new ArgumentOutOfRangeException(nameof(order), order.Volume, LocalizedStrings.Str895.Put(order.Price));
				}

				if (order.Type == null)
					order.Type = order.Price > 0 ? OrderTypes.Limit : OrderTypes.Market;

				InitNewOrder(order);

				OnRegisterOrder(order);
			}
			catch (Exception ex)
			{
				var transactionId = order.TransactionId;

				if (transactionId == 0 || order.State != OrderStates.None)
					transactionId = TransactionIdGenerator.GetNextId();

				SendOrderFailed(order, OrderOperations.Register, ex, transactionId);
			}
		}

		/// <inheritdoc />
		public bool? IsOrderEditable(Order order)
			=> _entityCache.TryGetAdapter(order)?.IsReplaceCommandEditCurrent;

		/// <inheritdoc />
		public bool? IsOrderReplaceable(Order order)
			=> _entityCache.TryGetAdapter(order)?.IsMessageSupported(MessageTypes.OrderReplace);

		/// <inheritdoc />
		public void EditOrder(Order order, Order changes)
		{
			if (order is null)
				throw new ArgumentNullException(nameof(order));

			if (changes is null)
				throw new ArgumentNullException(nameof(changes));

			try
			{
				this.AddOrderInfoLog(order, nameof(EditOrder));

				CheckOnOld(order);
				CheckOnNew(changes);

				if (IsOrderEditable(order) != true)
					this.AddWarningLog("Order {0} is not editable.", order.TransactionId);

				var transactionId = TransactionIdGenerator.GetNextId();
				_entityCache.AddOrderByEditionId(order, transactionId);

				changes.TransactionId = transactionId;
				OnEditOrder(order, changes);
			}
			catch (Exception ex)
			{
				SendOrderFailed(order, OrderOperations.Edit, ex, changes.TransactionId);
			}
		}

		/// <inheritdoc />
		public void ReRegisterOrder(Order oldOrder, Order newOrder)
		{
			if (oldOrder is null)
				throw new ArgumentNullException(nameof(oldOrder));

			if (newOrder is null)
				throw new ArgumentNullException(nameof(newOrder));

			try
			{
				this.AddOrderInfoLog(oldOrder, nameof(ReRegisterOrder));

				if (oldOrder.Security != newOrder.Security)
					throw new ArgumentException(LocalizedStrings.Str1098Params.Put(newOrder.Security.Id, oldOrder.Security.Id), nameof(newOrder));

				CheckOnOld(oldOrder);
				CheckOnNew(newOrder);

				if (IsOrderReplaceable(oldOrder) != true)
					this.AddWarningLog("Order {0} is not replaceable.", oldOrder.TransactionId);

				InitNewOrder(newOrder);
				_entityCache.AddOrderByCancelationId(oldOrder, newOrder.TransactionId);

				OnReRegisterOrder(oldOrder, newOrder);
			}
			catch (Exception ex)
			{
				var transactionId = newOrder.TransactionId;

				if (transactionId == 0 || newOrder.State != OrderStates.None)
					transactionId = TransactionIdGenerator.GetNextId();

				SendOrderFailed(oldOrder, OrderOperations.Cancel, ex, transactionId);
				SendOrderFailed(newOrder, OrderOperations.Register, ex, transactionId);
			}
		}

		/// <summary>
		/// Replace orders.
		/// </summary>
		/// <param name="oldOrder1">Cancelling order.</param>
		/// <param name="newOrder1">New order to register.</param>
		/// <param name="oldOrder2">Cancelling order.</param>
		/// <param name="newOrder2">New order to register.</param>
		public void ReRegisterOrderPair(Order oldOrder1, Order newOrder1, Order oldOrder2, Order newOrder2)
		{
			if (oldOrder1 == null)
				throw new ArgumentNullException(nameof(oldOrder1));

			if (newOrder1 == null)
				throw new ArgumentNullException(nameof(newOrder1));

			if (oldOrder2 == null)
				throw new ArgumentNullException(nameof(oldOrder2));

			if (newOrder2 == null)
				throw new ArgumentNullException(nameof(newOrder2));

			try
			{
				if (oldOrder1.Security != newOrder1.Security)
					throw new ArgumentException(LocalizedStrings.Str1099Params.Put(newOrder1.Security.Id, oldOrder1.Security.Id), nameof(newOrder1));

				if (oldOrder2.Security != newOrder2.Security)
					throw new ArgumentException(LocalizedStrings.Str1100Params.Put(newOrder2.Security.Id, oldOrder2.Security.Id), nameof(newOrder2));

				if (oldOrder1.Type == OrderTypes.Conditional || oldOrder2.Type == OrderTypes.Conditional)
				{
					CancelOrder(oldOrder1);
					RegisterOrder(newOrder1);

					CancelOrder(oldOrder2);
					RegisterOrder(newOrder2);
				}
				else
				{
					CheckOnOld(oldOrder1);
					CheckOnNew(newOrder1);

					CheckOnOld(oldOrder2);
					CheckOnNew(newOrder2);

					if (oldOrder1.Comment.IsEmpty())
						oldOrder1.Comment = newOrder1.Comment;

					if (oldOrder2.Comment.IsEmpty())
						oldOrder2.Comment = newOrder2.Comment;

					InitNewOrder(newOrder1);
					InitNewOrder(newOrder2);

					_entityCache.AddOrderByCancelationId(oldOrder1, newOrder1.TransactionId);
					_entityCache.AddOrderByCancelationId(oldOrder2, newOrder2.TransactionId);

					OnReRegisterOrderPair(oldOrder1, newOrder1, oldOrder2, newOrder2);
				}
			}
			catch (Exception ex)
			{
				var transactionId = newOrder1.TransactionId;

				if (transactionId == 0)
					transactionId = TransactionIdGenerator.GetNextId();

				SendOrderFailed(oldOrder1, OrderOperations.Cancel, ex, transactionId);
				SendOrderFailed(newOrder1, OrderOperations.Register, ex, transactionId);

				SendOrderFailed(oldOrder2, OrderOperations.Cancel, ex, transactionId);
				SendOrderFailed(newOrder2, OrderOperations.Register, ex, transactionId);
			}
		}

		/// <inheritdoc />
		public void CancelOrder(Order order)
		{
			long transactionId = 0;

			try
			{
				this.AddOrderInfoLog(order, nameof(CancelOrder));

				CheckOnOld(order);

				transactionId = TransactionIdGenerator.GetNextId();
				_entityCache.AddOrderByCancelationId(order, transactionId);

				OnCancelOrder(order, transactionId);
			}
			catch (Exception ex)
			{
				if (transactionId == 0)
					transactionId = TransactionIdGenerator.GetNextId();

				SendOrderFailed(order, OrderOperations.Cancel, ex, transactionId);
			}
		}

		private void SendOrderFailed(Order order, OrderOperations operation, Exception error, long originalTransactionId)
		{
			var fail = EntityFactory.CreateOrderFail(order, error);
			fail.ServerTime = CurrentTime;

			_entityCache.AddOrderFailById(fail, operation, originalTransactionId);

			SendOutMessage(fail.ToMessage(originalTransactionId));
		}

		private void CheckOnNew(Order order)
		{
			CheckOrderState(order);

			if (order.TransactionId != 0)
				throw new ArgumentException(LocalizedStrings.Str897Params.Put(order.TransactionId), nameof(order));

			if (order.State != OrderStates.None)
				throw new ArgumentException(LocalizedStrings.Str898Params.Put(order.State), nameof(order));

			if (order.Id != null || !order.StringId.IsEmpty())
				throw new ArgumentException(LocalizedStrings.Str896Params.Put(order.Id == null ? order.StringId : order.Id.To<string>()), nameof(order));

			if (CheckSteps)
			{
				if (order.Price > 0)
				{
					var priceStep = order.Security.PriceStep;

					if (priceStep != null && (order.Price % priceStep.Value) != 0)
						throw new ArgumentException(LocalizedStrings.OrderPriceNotMultipleOfPriceStep.Put(order.Price, order, priceStep.Value));
				}

				var volumeStep = order.Security.VolumeStep;

				if (volumeStep != null && (order.Volume % volumeStep.Value) != 0)
					throw new ArgumentException(LocalizedStrings.OrderVolumeNotMultipleOfVolumeStep.Put(order.Volume, order, volumeStep.Value));
			}
		}

		private static void CheckOnOld(Order order)
		{
			CheckOrderState(order);

			if (order.TransactionId == 0)
				throw new ArgumentException(LocalizedStrings.Str899, nameof(order));
		}

		private static void CheckOrderState(Order order)
		{
			if (order == null)
				throw new ArgumentNullException(nameof(order));

			if (order.Type == OrderTypes.Conditional && order.Condition == null)
				throw new ArgumentException(LocalizedStrings.Str889, nameof(order));

			if (order.Security == null)
				throw new ArgumentException(LocalizedStrings.Str890, nameof(order));

			if (order.Portfolio == null)
				throw new ArgumentException(LocalizedStrings.Str891, nameof(order));

			if (order.Price < 0)
				throw new ArgumentOutOfRangeException(nameof(order), order.Price, LocalizedStrings.Str892);

			if (order.Price == 0 && order.Type == OrderTypes.Limit)
				throw new ArgumentException(LocalizedStrings.Str893, nameof(order));
		}

		/// <summary>
		/// Initialize registering order (transaction id etc.).
		/// </summary>
		/// <param name="order">New order.</param>
		private void InitNewOrder(Order order)
		{
			order.Balance = order.Volume;

			//if (order.ExtensionInfo == null)
			//	order.ExtensionInfo = new Dictionary<string, object>();

			if (order.TransactionId == 0)
				order.TransactionId = TransactionIdGenerator.GetNextId();

			//order.Connector = this;

			//if (order.Security is ContinuousSecurity)
			//	order.Security = ((ContinuousSecurity)order.Security).GetSecurity(CurrentTime);

			order.LocalTime = CurrentTime;
			order.ApplyNewState(OrderStates.Pending, this);

			_entityCache.AddOrderByRegistrationId(order);

			SendOutMessage(order.ToMessage());
		}

		/// <summary>
		/// Register new order.
		/// </summary>
		/// <param name="order">Registration details.</param>
		protected void OnRegisterOrder(Order order)
		{
			SendInMessage(order.CreateRegisterMessage(GetSecurityId(order.Security)));
		}

		/// <summary>
		/// Edit the order.
		/// </summary>
		/// <param name="order">Order.</param>
		/// <param name="changes">Order changes.</param>
		protected void OnEditOrder(Order order, Order changes)
		{
			SendInMessage(order.CreateReplaceMessage(changes, GetSecurityId(order.Security)));
		}

		/// <summary>
		/// Reregister the order.
		/// </summary>
		/// <param name="oldOrder">Cancelling order.</param>
		/// <param name="newOrder">New order to register.</param>
		protected void OnReRegisterOrder(Order oldOrder, Order newOrder)
		{
			SendInMessage(oldOrder.CreateReplaceMessage(newOrder, GetSecurityId(newOrder.Security)));
		}

		/// <summary>
		/// Reregister of pair orders.
		/// </summary>
		/// <param name="oldOrder1">First order to cancel.</param>
		/// <param name="newOrder1">First new order to register.</param>
		/// <param name="oldOrder2">Second order to cancel.</param>
		/// <param name="newOrder2">Second new order to register.</param>
		protected void OnReRegisterOrderPair(Order oldOrder1, Order newOrder1, Order oldOrder2, Order newOrder2)
		{
			SendInMessage(oldOrder1.CreateReplaceMessage(newOrder1, GetSecurityId(newOrder1.Security), oldOrder2, newOrder2, GetSecurityId(newOrder2.Security)));
		}

		/// <summary>
		/// Cancel the order.
		/// </summary>
		/// <param name="order">Order to cancel.</param>
		/// <param name="transactionId">Order cancellation transaction id.</param>
		protected void OnCancelOrder(Order order, long transactionId)
		{
			SendInMessage(order.CreateCancelMessage(GetSecurityId(order.Security), transactionId));
		}

		/// <inheritdoc />
		public void CancelOrders(bool? isStopOrder = null, Portfolio portfolio = null, Sides? direction = null, ExchangeBoard board = null, Security security = null, SecurityTypes? securityType = null, long? transactionId = null)
		{
			if (transactionId == null)
				transactionId = TransactionIdGenerator.GetNextId();

			_entityCache.TryAddMassCancelationId(transactionId.Value);
			OnCancelOrders(transactionId.Value, isStopOrder, portfolio, direction, board, security, securityType);
		}

		/// <summary>
		/// Cancel orders by filter.
		/// </summary>
		/// <param name="transactionId">Order cancellation transaction id.</param>
		/// <param name="isStopOrder"><see langword="true" />, if cancel only a stop orders, <see langword="false" /> - if regular orders, <see langword="null" /> - both.</param>
		/// <param name="portfolio">Portfolio. If the value is equal to <see langword="null" />, then the portfolio does not match the orders cancel filter.</param>
		/// <param name="direction">Order side. If the value is <see langword="null" />, the direction does not use.</param>
		/// <param name="board">Trading board. If the value is equal to <see langword="null" />, then the board does not match the orders cancel filter.</param>
		/// <param name="security">Instrument. If the value is equal to <see langword="null" />, then the instrument does not match the orders cancel filter.</param>
		/// <param name="securityType">Security type. If the value is <see langword="null" />, the type does not use.</param>
		protected void OnCancelOrders(long transactionId, bool? isStopOrder = null, Portfolio portfolio = null, Sides? direction = null, ExchangeBoard board = null, Security security = null, SecurityTypes? securityType = null)
		{
			var cancelMsg = new OrderGroupCancelMessage
			{
				TransactionId = transactionId
			};

			if (security != null)
				cancelMsg.SecurityId = GetSecurityId(security);

			if (board != null)
			{
				var temp = cancelMsg.SecurityId;
				temp.BoardCode = board.Code;
				cancelMsg.SecurityId = temp;
			}

			if (portfolio != null)
				cancelMsg.PortfolioName = portfolio.Name;

			if (isStopOrder != null)
				cancelMsg.OrderType = isStopOrder == true ? OrderTypes.Conditional : OrderTypes.Limit;

			if (direction != null)
				cancelMsg.Side = direction.Value;

			//if (security != null)
			//	security.ToMessage(securityId).CopyTo(cancelMsg);

			if (securityType != null)
				cancelMsg.SecurityType = securityType;

			SendInMessage(cancelMsg);
		}

		/// <summary>
		/// Change password.
		/// </summary>
		/// <param name="newPassword">New password.</param>
		public void ChangePassword(SecureString newPassword)
		{
			var msg = new ChangePasswordMessage
			{
				NewPassword = newPassword,
				TransactionId = TransactionIdGenerator.GetNextId()
			};

			SendInMessage(msg);
		}

		private DateTimeOffset _prevTime;

		private void ProcessTimeInterval(Message message)
		{
			if (message == _marketTimeMessage)
			{
				lock (_marketTimerSync)
					_isMarketTimeHandled = true;
			}

			var currentTime = message.LocalTime;

			if (_prevTime == default)
			{
				_prevTime = currentTime;
				return;
			}

			var diff = currentTime - _prevTime;

			if (diff < MarketTimeChangedInterval)
				return;

			_prevTime = currentTime;
			RaiseMarketTimeChanged(diff);
		}

		/// <inheritdoc />
		public Security GetSecurity(SecurityId securityId)
			=> GetSecurity(securityId, s => false, out _);

		private Security TryGetSecurity(SecurityId? securityId)
			=> securityId == null || securityId.Value == default ? null : GetSecurity(securityId.Value);

		private Security EnsureGetSecurity<TMessage>(TMessage message)
			where TMessage : ISecurityIdMessage, ISubscriptionIdMessage
		{
			var secId = message.SecurityId;

			if (secId.IsMoney())
				return TraderHelper.MoneySecurity;

			if (secId == default)
			{
				var subscrSecId = message
					.GetSubscriptionIds()
					.Select(id => TryGetSubscriptionById(id)?.SecurityId)
					.Where(id => id != null && id.Value != default)
					.FirstOrDefault();

				if (subscrSecId == null || subscrSecId.Value == default)
					throw new ArgumentOutOfRangeException(nameof(message), message, LocalizedStrings.Str1025);

				secId = subscrSecId.Value;
			}

			var security = TryGetSecurity(secId);

			if (security == null)
				throw new ArgumentOutOfRangeException(nameof(message), message, LocalizedStrings.Str704Params.Put(secId));

			return security;
		}

		/// <summary>
		/// To get the instrument by the code. If the instrument is not found, then the <see cref="IEntityFactory.CreateSecurity"/> is called to create an instrument.
		/// </summary>
		/// <param name="id">Security ID.</param>
		/// <param name="changeSecurity">The handler changing the instrument. It returns <see langword="true" /> if the instrument has been changed and the <see cref="IConnector.SecuritiesChanged"/> should be called.</param>
		/// <param name="isNew">Is newly created.</param>
		/// <returns>Security.</returns>
		private Security GetSecurity(SecurityId id, Func<Security, bool> changeSecurity, out bool isNew)
		{
			if (id == default)
				throw new ArgumentNullException(nameof(id));

			if (changeSecurity == null)
				throw new ArgumentNullException(nameof(changeSecurity));

			var security = SecurityStorage.GetOrCreate(id, key =>
			{
				var s = EntityFactory.CreateSecurity(key);

				if (s == null)
					throw new InvalidOperationException(LocalizedStrings.Str1102Params.Put(key));

				var idInfo = SecurityIdGenerator.Split(key);

				var code = idInfo.SecurityCode;
				var board = ExchangeInfoProvider.GetOrCreateBoard(GetBoardCode(idInfo.BoardCode));

				if (s.Board == null)
					s.Board = board;

				if (s.Code.IsEmpty())
					s.Code = code;

				if (s.Name.IsEmpty())
					s.Name = code;

				//if (s.Class.IsEmpty())
				//	s.Class = board.Code;

				return s;
			}, out isNew);

			if (isNew)
				ExchangeInfoProvider.Save(security.Board);

			var isChanged = changeSecurity(security);

			if (_existingSecurities.TryAdd(security))
				RaiseNewSecurity(security);
			else if (isChanged)
				RaiseSecurityChanged(security);

			return security;
		}

		/// <inheritdoc />
		public SecurityId GetSecurityId(Security security)
			=> security.ToSecurityId(SecurityIdGenerator, copyExtended: true);

		private string GetBoardCode(string secClass)
			// MarketDataAdapter can be null then loading infos from StorageAdapter.
			=> MarketDataAdapter != null ? MarketDataAdapter.GetBoardCode(secClass) : secClass;

		/// <summary>
		/// Generate <see cref="Security.Id"/> security.
		/// </summary>
		/// <param name="secCode">Security code.</param>
		/// <param name="secClass">Security class.</param>
		/// <returns><see cref="Security.Id"/> security.</returns>
		protected string CreateSecurityId(string secCode, string secClass)
			=> SecurityIdGenerator.GenerateId(secCode, GetBoardCode(secClass));

		/// <inheritdoc />
		public object GetSecurityValue(Security security, Level1Fields field)
			=> _entityCache.GetSecurityValue(security, field);

		/// <inheritdoc />
		public IEnumerable<Level1Fields> GetLevel1Fields(Security security)
			=> _entityCache.GetLevel1Fields(security);

		/// <summary>
		/// Clear cache.
		/// </summary>
		public virtual void ClearCache()
		{
			_entityCache.Clear();

			_existingSecurities.Clear();
			_existingPortfolios.Clear();
			_existingPositions.Clear();

			_prevTime = default;

			ConnectionState = ConnectionStates.Disconnected;

			_subscriptionManager.ClearCache();

			SendInMessage(new ResetMessage());

			CloseTimer();

			_cleared?.Invoke();
		}

		/// <summary>
		/// To release allocated resources. In particular, to disconnect from the trading system via <see cref="Connector.Disconnect"/>.
		/// </summary>
		protected override void DisposeManaged()
		{
			_isDisposing = true;

			if (ConnectionState == ConnectionStates.Connected)
			{
				try
				{
					Disconnect();
				}
				catch (Exception ex)
				{
					RaiseConnectionError(ex);
				}
			}

			base.DisposeManaged();

			//if (ConnectionState == ConnectionStates.Disconnected || ConnectionState == ConnectionStates.Failed)
			//	TransactionAdapter = null;

			//if (ExportState == ConnectionStates.Disconnected || ExportState == ConnectionStates.Failed)
			//	MarketDataAdapter = null;

			SendInMessage(_disposeMessage);

			CloseTimer();
		}

		/// <inheritdoc />
		public override void Load(SettingsStorage storage)
		{
			if (storage is null)
				throw new ArgumentNullException(nameof(storage));

			TradesKeepCount = storage.GetValue(nameof(TradesKeepCount), TradesKeepCount);
			OrdersKeepCount = storage.GetValue(nameof(OrdersKeepCount), OrdersKeepCount);
			UpdateSecurityLastQuotes = storage.GetValue(nameof(UpdateSecurityLastQuotes), UpdateSecurityLastQuotes);
			UpdateSecurityByLevel1 = storage.GetValue(nameof(UpdateSecurityByLevel1), UpdateSecurityByLevel1);
			UpdateSecurityByDefinition = storage.GetValue(nameof(UpdateSecurityByDefinition), UpdateSecurityByDefinition);
			//ReConnectionSettings.Load(storage.GetValue<SettingsStorage>(nameof(ReConnectionSettings)));
			OverrideSecurityData = storage.GetValue(nameof(OverrideSecurityData), OverrideSecurityData);
			CheckSteps = storage.GetValue(nameof(CheckSteps), CheckSteps);
			KeepStrategiesPositions = storage.GetValue(nameof(KeepStrategiesPositions), KeepStrategiesPositions);

			if (storage.ContainsKey(nameof(RiskManager)))
				RiskManager = storage.GetValue<SettingsStorage>(nameof(RiskManager)).LoadEntire<IRiskManager>();

			Adapter.Load(storage.GetValue<SettingsStorage>(nameof(Adapter)));

			MarketTimeChangedInterval = storage.GetValue<TimeSpan>(nameof(MarketTimeChangedInterval));
			SupportAssociatedSecurity = storage.GetValue(nameof(SupportAssociatedSecurity), SupportAssociatedSecurity);

			var lookupMessagesOnConnect = storage.GetValue<object>(nameof(LookupMessagesOnConnect));
			if (lookupMessagesOnConnect is bool b)
			{
				if (!b)
					LookupMessagesOnConnect.Clear();
			}
			else if (lookupMessagesOnConnect is string str)
			{
				LookupMessagesOnConnect.Clear();
				LookupMessagesOnConnect.AddRange(str.SplitByComma(true).Select(s => s.To<MessageTypes>()));
			}

			IsRestoreSubscriptionOnNormalReconnect = storage.GetValue(nameof(IsRestoreSubscriptionOnNormalReconnect), IsRestoreSubscriptionOnNormalReconnect);
			IsAutoUnSubscribeOnDisconnect = storage.GetValue(nameof(IsAutoUnSubscribeOnDisconnect), IsAutoUnSubscribeOnDisconnect);
			IsAutoPortfoliosSubscribe = storage.GetValue(nameof(IsAutoPortfoliosSubscribe), IsAutoPortfoliosSubscribe);

			if (Buffer != null && storage.ContainsKey(nameof(Buffer)))
				Buffer.ForceLoad(storage.GetValue<SettingsStorage>(nameof(Buffer)));

			base.Load(storage);
		}

		/// <inheritdoc />
		public override void Save(SettingsStorage storage)
		{
			if (storage is null)
				throw new ArgumentNullException(nameof(storage));

			storage.SetValue(nameof(TradesKeepCount), TradesKeepCount);
			storage.SetValue(nameof(OrdersKeepCount), OrdersKeepCount);
			storage.SetValue(nameof(UpdateSecurityLastQuotes), UpdateSecurityLastQuotes);
			storage.SetValue(nameof(UpdateSecurityByLevel1), UpdateSecurityByLevel1);
			storage.SetValue(nameof(UpdateSecurityByDefinition), UpdateSecurityByDefinition);
			//storage.SetValue(nameof(ReConnectionSettings), ReConnectionSettings.Save());
			storage.SetValue(nameof(OverrideSecurityData), OverrideSecurityData);
			storage.SetValue(nameof(CheckSteps), CheckSteps);
			storage.SetValue(nameof(KeepStrategiesPositions), KeepStrategiesPositions);

			if (RiskManager != null)
				storage.SetValue(nameof(RiskManager), RiskManager.SaveEntire(false));

			storage.SetValue(nameof(Adapter), Adapter.Save());

			storage.SetValue(nameof(MarketTimeChangedInterval), MarketTimeChangedInterval);
			storage.SetValue(nameof(SupportAssociatedSecurity), SupportAssociatedSecurity);

			storage.SetValue(nameof(LookupMessagesOnConnect), _lookupMessagesOnConnect.Cache.Select(t => t.To<string>()).JoinComma());
			storage.SetValue(nameof(IsRestoreSubscriptionOnNormalReconnect), IsRestoreSubscriptionOnNormalReconnect);
			storage.SetValue(nameof(IsAutoUnSubscribeOnDisconnect), IsAutoUnSubscribeOnDisconnect);
			storage.SetValue(nameof(IsAutoPortfoliosSubscribe), IsAutoPortfoliosSubscribe);

			if (Buffer != null)
				storage.SetValue(nameof(Buffer), Buffer.Save());

			base.Save(storage);
		}

		#region ICandleManager implementation

		int ICandleSource<Candle>.SpeedPriority => 0;

		event Action<CandleSeries, Candle> ICandleSource<Candle>.Processing
		{
			add => CandleSeriesProcessing += value;
			remove => CandleSeriesProcessing -= value;
		}

		event Action<CandleSeries> ICandleSource<Candle>.Stopped
		{
			add => CandleSeriesStopped += value;
			remove => CandleSeriesStopped -= value;
		}

		IEnumerable<Range<DateTimeOffset>> ICandleSource<Candle>.GetSupportedRanges(CandleSeries series)
			=> Enumerable.Empty<Range<DateTimeOffset>>();

		void ICandleSource<Candle>.Start(CandleSeries series, DateTimeOffset? from, DateTimeOffset? to)
			=> this.SubscribeCandles(series, from, to);

		void ICandleSource<Candle>.Stop(CandleSeries series)
#pragma warning disable CS0618 // Type or member is obsolete
			=> this.UnSubscribeCandles(series);
#pragma warning restore CS0618 // Type or member is obsolete

		ICandleManagerContainer ICandleManager.Container { get; } = new CandleManagerContainer();

		IEnumerable<CandleSeries> ICandleManager.Series => SubscribedCandleSeries;

		IList<ICandleSource<Candle>> ICandleManager.Sources => Array.Empty<ICandleSource<Candle>>();

		#endregion

		#region IMessageChannel implementation

		private Action<Message> _newOutMessage;

		event Action<Message> IMessageChannel.NewOutMessage
		{
			add => _newOutMessage += value;
			remove => _newOutMessage -= value;
		}

		ChannelStates IMessageChannel.State => ConnectionState == ConnectionStates.Connected ? ChannelStates.Started : ChannelStates.Stopped;

		private Action _stateChanged;

		event Action IMessageChannel.StateChanged
		{
			add => _stateChanged += value;
			remove => _stateChanged -= value;
		}

		void IMessageChannel.Open()
		{
			Connect();
		}

		void IMessageChannel.Close()
		{
			Disconnect();
		}

		void IMessageChannel.Suspend()
		{
		}

		void IMessageChannel.Resume()
		{
		}

		void IMessageChannel.Clear()
		{
		}

		IMessageChannel ICloneable<IMessageChannel>.Clone()
		{
			return this.Clone();
		}

		object ICloneable.Clone()
		{
			return this.Clone();
		}

		#endregion
	}
}

redpajama_set_name: RedPajamaGithub


text:
package org.elasticsearch.action;

import com.google.common.base.Preconditions;
import org.elasticsearch.action.support.PlainListenableActionFuture;
import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.threadpool.ThreadPool;

public abstract class ActionRequestBuilder<Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder>> {

    protected final Action<Request, Response, RequestBuilder> action;
    protected final Request request;
    private final ThreadPool threadPool;
    protected final ElasticsearchClient client;

    protected ActionRequestBuilder(ElasticsearchClient client, Action<Request, Response, RequestBuilder> action, Request request) {
        Preconditions.checkNotNull(action, "action must not be null");
        this.action = action;
        this.request = request;
        this.client = client;
        threadPool = client.threadPool();
    }

    public Request request() {
        return this.request;
    }

    @SuppressWarnings("unchecked")
    public final RequestBuilder putHeader(String key, Object value) {
        request.putHeader(key, value);
        return (RequestBuilder) this;
    }

    public ListenableActionFuture<Response> execute() {
        PlainListenableActionFuture<Response> future = new PlainListenableActionFuture<>(threadPool);
        execute(future);
        return future;
    }

    /**
     * Short version of execute().actionGet().
     */
    public Response get() {
        return execute().actionGet();
    }

    /**
     * Short version of execute().actionGet().
     *
     * @param timeout Timeout
     */
    public Response get(TimeValue timeout) {
        return execute().actionGet(timeout);
    }

    /**
     * Short version of execute().actionGet().
     *
     * @param timeout Timeout
     */
    public Response get(String timeout) {
        return execute().actionGet(timeout);
    }

    public void execute(ActionListener<Response> listener) {
        client.execute(action, beforeExecute(request), listener);
    }

    /**
     * A callback to additionally process the request before its executed
     *
     * @param request Request
     */
    protected Request beforeExecute(Request request) {
        return request;
    }
}


text:
package com.sequenceiq.cloudbreak.cmtemplate.configproviders.ssb;

import static com.sequenceiq.cloudbreak.cmtemplate.configproviders.ConfigUtils.config;
import static java.util.Collections.emptyList;

import java.util.List;

import org.springframework.stereotype.Component;

import com.cloudera.api.swagger.model.ApiClusterTemplateConfig;
import com.sequenceiq.cloudbreak.api.endpoint.v4.database.base.DatabaseType;
import com.sequenceiq.cloudbreak.template.TemplatePreparationObject;
import com.sequenceiq.cloudbreak.template.views.RdsView;

@Component
public class SqlStreamBuilderMveDatabaseConfigProvider extends SqlStreamBuilderConfigProvider {

    static final String DATABASE_URL = "ssb.mve.datasource.url";

    static final String DATABASE_USER = "ssb.mve.datasource.username";

    static final String DATABASE_PASSWORD = "ssb.mve.datasource.password";

    @Override
    public String getServiceType() {
        return SqlStreamBuilderRoles.SQL_STREAM_BUILDER;
    }

    @Override
    public List<String> getRoleTypes() {
        return List.of(SqlStreamBuilderRoles.MATERIALIZED_VIEW_ENGINE);
    }

    @Override
    protected List<ApiClusterTemplateConfig> getRoleConfigs(String roleType, TemplatePreparationObject source) {
        RdsView ssbRdsView = getRdsView(source);

        if (SqlStreamBuilderRoles.MATERIALIZED_VIEW_ENGINE.equals(roleType)) {
            return List.of(
                    config(DATABASE_URL, ssbRdsView.getConnectionURL()),
                    config(DATABASE_USER, ssbRdsView.getConnectionUserName()),
                    config(DATABASE_PASSWORD, ssbRdsView.getConnectionPassword())
            );
        }

        return emptyList();
    }

    @Override
    protected DatabaseType dbType() {
        return DatabaseType.SQL_STREAM_BUILDER_SNAPPER;
    }
}


text:
package com.intellij.vcs.log.impl;

import com.intellij.openapi.Disposable;
import com.intellij.openapi.diagnostic.Logger;
import com.intellij.openapi.extensions.ExtensionPointName;
import com.intellij.openapi.project.Project;
import com.intellij.openapi.util.Disposer;
import com.intellij.openapi.vcs.AbstractVcs;
import com.intellij.openapi.vcs.VcsRoot;
import com.intellij.openapi.vcs.changes.ui.ChangesViewContentManager;
import com.intellij.openapi.vfs.VirtualFile;
import com.intellij.openapi.wm.IdeFocusManager;
import com.intellij.openapi.wm.ToolWindowManager;
import com.intellij.openapi.wm.ex.ToolWindowManagerListener;
import com.intellij.openapi.wm.impl.ToolWindowImpl;
import com.intellij.openapi.wm.impl.ToolWindowManagerImpl;
import com.intellij.ui.content.Content;
import com.intellij.ui.content.ContentManagerAdapter;
import com.intellij.ui.content.ContentManagerEvent;
import com.intellij.util.Consumer;
import com.intellij.util.containers.ContainerUtil;
import com.intellij.util.containers.MultiMap;
import com.intellij.util.ui.UIUtil;
import com.intellij.vcs.log.VcsLogProvider;
import com.intellij.vcs.log.VcsLogRefresher;
import com.intellij.vcs.log.VcsLogSettings;
import com.intellij.vcs.log.data.VcsLogDataHolder;
import com.intellij.vcs.log.data.VcsLogUiProperties;
import com.intellij.vcs.log.data.VisiblePack;
import com.intellij.vcs.log.ui.VcsLogColorManagerImpl;
import com.intellij.vcs.log.ui.VcsLogUiImpl;
import com.intellij.vcs.log.ui.frame.VcsLogGraphTable;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;

import javax.swing.*;
import java.util.*;

public class VcsLogManager implements Disposable {

  public static final ExtensionPointName<VcsLogProvider> LOG_PROVIDER_EP = ExtensionPointName.create("com.intellij.logProvider");
  private static final Logger LOG = Logger.getInstance(VcsLogManager.class);

  @NotNull private final Project myProject;
  @NotNull private final VcsLogSettings mySettings;
  @NotNull private final VcsLogUiProperties myUiProperties;

  private VcsLogRefresher myLogRefresher;
  private volatile VcsLogUiImpl myUi;

  public VcsLogManager(@NotNull Project project,
                       @NotNull VcsLogSettings settings,
                       @NotNull VcsLogUiProperties uiProperties) {
    myProject = project;
    mySettings = settings;
    myUiProperties = uiProperties;
  }

  @NotNull
  public JComponent initContent(@NotNull Collection<VcsRoot> roots, @Nullable String contentTabName) {
    Disposer.register(myProject, this);

    final Map<VirtualFile, VcsLogProvider> logProviders = findLogProviders(roots, myProject);

    Consumer<VisiblePack> visiblePackConsumer = new Consumer<VisiblePack>() {
      @Override
      public void consume(final VisiblePack pack) {
        UIUtil.invokeLaterIfNeeded(new Runnable() {
          @Override
          public void run() {
            if (!Disposer.isDisposed(myUi)) {
              myUi.setVisiblePack(pack);
            }
          }
        });
      }
    };
    final VcsLogDataHolder logDataHolder = new VcsLogDataHolder(myProject, this, logProviders, mySettings, myUiProperties, visiblePackConsumer);
    myUi = new VcsLogUiImpl(logDataHolder, myProject, mySettings,
                            new VcsLogColorManagerImpl(logProviders.keySet()), myUiProperties, logDataHolder.getFilterer());
    myUi.addLogListener(logDataHolder.getContainingBranchesGetter()); // TODO: remove this after VcsLogDataHolder vs VcsLoUi dependency cycle is solved
    if (contentTabName != null) {
      myLogRefresher = new PostponeableLogRefresher(myProject, logDataHolder, contentTabName);
    }
    else {
      myLogRefresher = new VcsLogRefresher() {
        @Override
        public void refresh(@NotNull VirtualFile root) {
          logDataHolder.refresh(Collections.singletonList(root));
        }
      };
    }
    refreshLogOnVcsEvents(logProviders);
    logDataHolder.initialize();

    // todo fix selection
    final VcsLogGraphTable graphTable = myUi.getTable();
    if (graphTable.getRowCount() > 0) {
      IdeFocusManager.getInstance(myProject).requestFocus(graphTable, true).doWhenProcessed(new Runnable() {
        @Override
        public void run() {
          graphTable.setRowSelectionInterval(0, 0);
        }
      });
    }
    return myUi.getMainFrame().getMainComponent();
  }

  private void refreshLogOnVcsEvents(@NotNull Map<VirtualFile, VcsLogProvider> logProviders) {
    MultiMap<VcsLogProvider, VirtualFile> providers2roots = MultiMap.create();
    for (Map.Entry<VirtualFile, VcsLogProvider> entry : logProviders.entrySet()) {
      providers2roots.putValue(entry.getValue(), entry.getKey());
    }

    for (Map.Entry<VcsLogProvider, Collection<VirtualFile>> entry : providers2roots.entrySet()) {
      entry.getKey().subscribeToRootRefreshEvents(entry.getValue(), myLogRefresher);
    }
  }

  @NotNull
  public static Map<VirtualFile, VcsLogProvider> findLogProviders(@NotNull Collection<VcsRoot> roots, @NotNull Project project) {
    Map<VirtualFile, VcsLogProvider> logProviders = ContainerUtil.newHashMap();
    VcsLogProvider[] allLogProviders = LOG_PROVIDER_EP.getExtensions(project);
    for (VcsRoot root : roots) {
      AbstractVcs vcs = root.getVcs();
      VirtualFile path = root.getPath();
      if (vcs == null || path == null) {
        LOG.error("Skipping invalid VCS root: " + root);
        continue;
      }

      for (VcsLogProvider provider : allLogProviders) {
        if (provider.getSupportedVcs().equals(vcs.getKeyInstanceMethod())) {
          logProviders.put(path, provider);
          break;
        }
      }
    }
    return logProviders;
  }

  /**
   * The instance of the {@link com.intellij.vcs.log.ui.VcsLogUiImpl} or null if the log was not initialized yet.
   */
  @Nullable
  public VcsLogUiImpl getLogUi() {
    return myUi;
  }

  @Override
  public void dispose() {
  }

  private static class PostponeableLogRefresher implements VcsLogRefresher, Disposable {

    private static final String TOOLWINDOW_ID = ChangesViewContentManager.TOOLWINDOW_ID;

    @NotNull private final VcsLogDataHolder myDataHolder;
    @NotNull private final ToolWindowManagerImpl myToolWindowManager;
    @NotNull private final ToolWindowImpl myToolWindow;
    @NotNull private final String myTabName;
    @NotNull private final MyRefreshPostponedEventsListener myPostponedEventsListener;

    @NotNull private final Set<VirtualFile> myRootsToRefresh = ContainerUtil.newConcurrentSet();

    public PostponeableLogRefresher(@NotNull Project project, @NotNull VcsLogDataHolder dataHolder, @NotNull String contentTabName) {
      myDataHolder = dataHolder;
      myToolWindowManager = (ToolWindowManagerImpl)ToolWindowManager.getInstance(project);
      myToolWindow = (ToolWindowImpl)myToolWindowManager.getToolWindow(TOOLWINDOW_ID);
      myTabName = contentTabName;

      Disposer.register(myToolWindow.getContentManager(), this);

      myPostponedEventsListener = new MyRefreshPostponedEventsListener();
      myToolWindow.getContentManager().addContentManagerListener(myPostponedEventsListener);
      myToolWindowManager.addToolWindowManagerListener(myPostponedEventsListener);
    }

    @Override
    public void refresh(@NotNull VirtualFile root) {
      if (isOurContentPaneShowing()) {
        myDataHolder.refresh(Collections.singleton(root));
      }
      else {
        myRootsToRefresh.add(root);
      }
    }

    @Override
    public void dispose() {
      myToolWindow.getContentManager().removeContentManagerListener(myPostponedEventsListener);
      myToolWindowManager.removeToolWindowManagerListener(myPostponedEventsListener);
    }

    private boolean isOurContentPaneShowing() {
      if (myToolWindowManager.isToolWindowRegistered(TOOLWINDOW_ID) && myToolWindow.isVisible()) {
        Content content = myToolWindow.getContentManager().getSelectedContent();
        return content != null && content.getTabName().equals(myTabName);
      }
      return false;
    }

    private void refreshPostponedRoots() {
      Set<VirtualFile> toRefresh = new HashSet<VirtualFile>(myRootsToRefresh);
      myRootsToRefresh.removeAll(toRefresh); // clear the set, but keep roots which could possibly arrive after collecting them in the var.
      myDataHolder.refresh(toRefresh);
    }

    private class MyRefreshPostponedEventsListener extends ContentManagerAdapter implements ToolWindowManagerListener {

      @Override
      public void selectionChanged(ContentManagerEvent event) {
        refreshRootsIfNeeded();
      }

      @Override
      public void stateChanged() {
        refreshRootsIfNeeded();
      }

      @Override
      public void toolWindowRegistered(@NotNull String id) {
      }

      private void refreshRootsIfNeeded() {
        if (isOurContentPaneShowing()) {
          refreshPostponedRoots();
        }
      }
    }
  }

}


text:
// Generated by the protocol buffer compiler.  DO NOT EDIT!
// source: google/type/timeofday.proto

package com.google.type;

public interface TimeOfDayOrBuilder extends
    // @@protoc_insertion_point(interface_extends:google.type.TimeOfDay)
    com.google.protobuf.MessageOrBuilder {

  /**
   *

   * Hours of day in 24 hour format. Should be from 0 to 23. An API may choose
   * to allow the value "24:00:00" for scenarios like business closing time.
   *

   *
   * int32 hours = 1;
   */
  int getHours();

  /**
   *

   * Minutes of hour of day. Must be from 0 to 59.
   *

   *
   * int32 minutes = 2;
   */
  int getMinutes();

  /**
   *

   * Seconds of minutes of the time. Must normally be from 0 to 59. An API may
   * allow the value 60 if it allows leap-seconds.
   *

   *
   * int32 seconds = 3;
   */
  int getSeconds();

  /**
   *

   * Fractions of seconds in nanoseconds. Must be from 0 to 999,999,999.
   *

   *
   * int32 nanos = 4;
   */
  int getNanos();
}