text:
package v1alpha1

import (
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"reflect"
	"sort"
	"strings"
	"time"

	"github.com/Masterminds/semver/v3"
	"github.com/imdario/mergo"
	"github.com/spf13/cast"
	v1 "k8s.io/api/core/v1"
	netv1 "k8s.io/api/networking/v1"
	extv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/pointer"
	ctrl "sigs.k8s.io/controller-runtime"
)

var log = ctrl.Log.WithName("controller_vault")

var bankVaultsImage string

func init() {
	if bankVaultsImage = os.Getenv("BANK_VAULTS_IMAGE"); bankVaultsImage == "" {
		bankVaultsImage = "ghcr.io/banzaicloud/bank-vaults:latest"
	}
}

// Vault is the Schema for the vaults API

// +genclient
// +genclient:noStatus
// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
// +k8s:openapi-gen=true
type Vault struct {
	metav1.TypeMeta   `json:",inline"`
	metav1.ObjectMeta `json:"metadata,omitempty"`

	Spec   VaultSpec   `json:"spec,omitempty"`
	Status VaultStatus `json:"status,omitempty"`
}

// VaultList contains a list of Vault

// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
type VaultList struct {
	metav1.TypeMeta `json:",inline"`
	metav1.ListMeta `json:"metadata,omitempty"`
	Items           []Vault `json:"items"`
}

// NOTE: json tags are required.  Any new fields you add must have json tags for the fields to be serialized.

// VaultSpec defines the desired state of Vault
// Important: Run "make generate-code" to regenerate code after modifying this file
type VaultSpec struct {

	// Size defines the number of Vault instances in the cluster (>= 1 means HA)
	// default: 1
	Size int32 `json:"size,omitempty"`

	// Image specifies the Vault image to use for the Vault instances
	// default: library/vault:latest
	Image string `json:"image,omitempty"`

	// BankVaultsImage specifies the Bank Vaults image to use for Vault unsealing and configuration
	// default: ghcr.io/banzaicloud/bank-vaults:latest
	BankVaultsImage string `json:"bankVaultsImage,omitempty"`

	// BankVaultsVolumeMounts define some extra Kubernetes Volume mounts for the Bank Vaults Sidecar container.
	// default:
	BankVaultsVolumeMounts []v1.VolumeMount `json:"bankVaultsVolumeMounts,omitempty"`

	// StatsDDisabled specifies if StatsD based metrics should be disabled
	// default: false
	StatsDDisabled bool `json:"statsdDisabled,omitempty"`

	// StatsDImage specifices the StatsD image to use for Vault metrics exportation
	// default: prom/statsd-exporter:latest
	StatsDImage string `json:"statsdImage,omitempty"`

	// StatsdConfig specifices the StatsD mapping configuration
	// default:
	StatsdConfig string `json:"statsdConfig,omitempty"`

	// FluentDEnabled specifies if FluentD based log exportation should be enabled
	// default: false
	FluentDEnabled bool `json:"fluentdEnabled,omitempty"`

	// FluentDImage specifices the FluentD image to use for Vault log exportation
	// default: fluent/fluentd:edge
	FluentDImage string `json:"fluentdImage,omitempty"`

	// FluentDConfLocation is the location of the fluent.conf file
	// default: "/fluentd/etc"
	FluentDConfLocation string `json:"fluentdConfLocation,omitempty"`

	// FluentDConfFile specifices the FluentD configuration file name to use for Vault log exportation
	// default:
	FluentDConfFile string `json:"fluentdConfFile,omitempty"`

	// FluentDConfig specifices the FluentD configuration to use for Vault log exportation
	// default:
	FluentDConfig string `json:"fluentdConfig,omitempty"`

	// WatchedSecretsLabels specifices a set of Kubernetes label selectors which select Secrets to watch.
	// If these Secrets change the Vault cluster gets restarted. For example a Secret that Cert-Manager is
	// managing a public Certificate for Vault using let's Encrypt.
	// default:
	WatchedSecretsLabels []map[string]string `json:"watchedSecretsLabels,omitempty"`

	// WatchedSecretsAnnotations specifices a set of Kubernetes annotations selectors which select Secrets to watch.
	// If these Secrets change the Vault cluster gets restarted. For example a Secret that Cert-Manager is
	// managing a public Certificate for Vault using let's Encrypt.
	// default:
	WatchedSecretsAnnotations []map[string]string `json:"watchedSecretsAnnotations,omitempty"`

	// Annotations define a set of common Kubernetes annotations that will be added to all operator managed resources.
	// default:
	Annotations map[string]string `json:"annotations,omitempty"`

	// VaultAnnotations define a set of Kubernetes annotations that will be added to all Vault Pods.
	// default:
	VaultAnnotations map[string]string `json:"vaultAnnotations,omitempty"`

	// VaultLabels define a set of Kubernetes labels that will be added to all Vault Pods.
	// default:
	VaultLabels map[string]string `json:"vaultLabels,omitempty"`

	// VaultPodSpec is a Kubernetes Pod specification snippet (`spec:` block) that will be merged into the operator generated
	// Vault Pod specification.
	// default:
	VaultPodSpec *EmbeddedPodSpec `json:"vaultPodSpec,omitempty"`

	// VaultContainerSpec is a Kubernetes Container specification snippet that will be merged into the operator generated
	// Vault Container specification.
	// default:
	VaultContainerSpec v1.Container `json:"vaultContainerSpec,omitempty"`

	// VaultConfigurerAnnotations define a set of Kubernetes annotations that will be added to the Vault Configurer Pod.
	// default:
	VaultConfigurerAnnotations map[string]string `json:"vaultConfigurerAnnotations,omitempty"`

	// VaultConfigurerLabels define a set of Kubernetes labels that will be added to all Vault Configurer Pod.
	// default:
	VaultConfigurerLabels map[string]string `json:"vaultConfigurerLabels,omitempty"`

	// VaultConfigurerPodSpec is a Kubernetes Pod specification snippet (`spec:` block) that will be merged into
	// the operator generated Vault Configurer Pod specification.
	// default:
	VaultConfigurerPodSpec *EmbeddedPodSpec `json:"vaultConfigurerPodSpec,omitempty"`

	// Config is the Vault Server configuration. See https://www.vaultproject.io/docs/configuration/ for more details.
	// default:
	Config extv1beta1.JSON `json:"config"`

	// ExternalConfig is higher level configuration block which instructs the Bank Vaults Configurer to configure Vault
	// through its API, thus allows setting up:
	// - Secret Engines
	// - Auth Methods
	// - Audit Devices
	// - Plugin Backends
	// - Policies
	// - Startup Secrets (Bank Vaults feature)
	// A documented example: https://github.com/banzaicloud/bank-vaults/blob/main/vault-config.yml
	// default:
	ExternalConfig extv1beta1.JSON `json:"externalConfig,omitempty"`

	// UnsealConfig defines where the Vault cluster's unseal keys and root token should be stored after initialization.
	// See the type's documentation for more details. Only one method may be specified.
	// default: Kubernetes Secret based unsealing
	UnsealConfig UnsealConfig `json:"unsealConfig,omitempty"`

	// CredentialsConfig defines a external Secret for Vault and how it should be mounted to the Vault Pod
	// for example accessing Cloud resources.
	// default:
	CredentialsConfig CredentialsConfig `json:"credentialsConfig,omitempty"`

	// EnvsConfig is a list of Kubernetes environment variable definitions that will be passed to all Bank-Vaults pods.
	// default:
	EnvsConfig []v1.EnvVar `json:"envsConfig,omitempty"`

	// SecurityContext is a Kubernetes PodSecurityContext that will be applied to all Pods created by the operator.
	// default:
	SecurityContext v1.PodSecurityContext `json:"securityContext,omitempty"`

	// ServiceType is a Kubernetes Service type of the Vault Service.
	// default: ClusterIP
	ServiceType string `json:"serviceType,omitempty"`

	// LoadBalancerIP is an optional setting for allocating a specific address for the entry service object
	// of type LoadBalancer
	// default: ""
	LoadBalancerIP string `json:"loadBalancerIP,omitempty"`

	// serviceRegistrationEnabled enables the injection of the service_registration Vault stanza.
	// This requires elaborated RBAC privileges for updating Pod labels for the Vault Pod.
	// default: false
	ServiceRegistrationEnabled bool `json:"serviceRegistrationEnabled,omitempty"`

	// RaftLeaderAddress defines the leader address of the raft cluster in multi-cluster deployments.
	// (In single cluster (namespace) deployments it is automatically detected).
	// "self" is a special value which means that this instance should be the bootstrap leader instance.
	// default: ""
	RaftLeaderAddress string `json:"raftLeaderAddress,omitempty"`

	// ServicePorts is an extra map of ports that should be exposed by the Vault Service.
	// default:
	ServicePorts map[string]int32 `json:"servicePorts,omitempty"`

	// Affinity is a group of affinity scheduling rules applied to all Vault Pods.
	// default:
	Affinity *v1.Affinity `json:"affinity,omitempty"`

	// PodAntiAffinity is the TopologyKey in the Vault Pod's PodAntiAffinity.
	// No PodAntiAffinity is used if empty.
	// Deprecated. Use Affinity.
	// default:
	PodAntiAffinity string `json:"podAntiAffinity,omitempty"`

	// NodeAffinity is Kubernetees NodeAffinity definition that should be applied to all Vault Pods.
	// Deprecated. Use Affinity.
	// default:
	NodeAffinity v1.NodeAffinity `json:"nodeAffinity,omitempty"`

	// NodeSelector is Kubernetees NodeSelector definition that should be applied to all Vault Pods.
	// default:
	NodeSelector map[string]string `json:"nodeSelector,omitempty"`

	// Tolerations is Kubernetes Tolerations definition that should be applied to all Vault Pods.
	// default:
	Tolerations []v1.Toleration `json:"tolerations,omitempty"`

	// ServiceAccount is Kubernetes ServiceAccount in which the Vault Pods should be running in.
	// default: default
	ServiceAccount string `json:"serviceAccount,omitempty"`

	// Volumes define some extra Kubernetes Volumes for the Vault Pods.
	// default:
	Volumes []v1.Volume `json:"volumes,omitempty"`

	// VolumeMounts define some extra Kubernetes Volume mounts for the Vault Pods.
	// default:
	VolumeMounts []v1.VolumeMount `json:"volumeMounts,omitempty"`

	// VolumeClaimTemplates define some extra Kubernetes PersistentVolumeClaim templates for the Vault Statefulset.
	// default:
	VolumeClaimTemplates []EmbeddedPersistentVolumeClaim `json:"volumeClaimTemplates,omitempty"`

	// VaultEnvsConfig is a list of Kubernetes environment variable definitions that will be passed to the Vault container.
	// default:
	VaultEnvsConfig []v1.EnvVar `json:"vaultEnvsConfig,omitempty"`

	// SidecarEnvsConfig is a list of Kubernetes environment variable definitions that will be passed to Vault sidecar containers.
	// default:
	SidecarEnvsConfig []v1.EnvVar `json:"sidecarEnvsConfig,omitempty"`

	// Resources defines the resource limits for all the resources created by the operator.
	// See the type for more details.
	// default:
	Resources *Resources `json:"resources,omitempty"`

	// Ingress, if it is specified the operator will create an Ingress resource for the Vault Service and
	// will annotate it with the correct Ingress annotations specific to the TLS settings in the configuration.
	// See the type for more details.
	// default:
	Ingress *Ingress `json:"ingress,omitempty"`

	// ServiceMonitorEnabled enables the creation of Prometheus Operator specific ServiceMonitor for Vault.
	// default: false
	ServiceMonitorEnabled bool `json:"serviceMonitorEnabled,omitempty"`

	// ExistingTLSSecretName is name of the secret that contains a TLS server certificate and key and the corresponding CA certificate.
	// Required secret format kubernetes.io/tls type secret keys + ca.crt key
	// If it is set, generating certificate will be disabled
	// default: ""
	ExistingTLSSecretName string `json:"existingTlsSecretName,omitempty"`

	// TLSExpiryThreshold is the Vault TLS certificate expiration threshold in Go's Duration format.
	// default: 168h
	TLSExpiryThreshold string `json:"tlsExpiryThreshold,omitempty"`

	// TLSAdditionalHosts is a list of additional hostnames or IP addresses to add to the SAN on the automatically generated TLS certificate.
	// default:
	TLSAdditionalHosts []string `json:"tlsAdditionalHosts,omitempty"`

	// CANamespaces define a list of namespaces where the generated CA certificate for Vault should be distributed,
	// use ["*"] for all namespaces.
	// default:
	CANamespaces []string `json:"caNamespaces,omitempty"`

	// IstioEnabled describes if the cluster has a Istio running and enabled.
	// default: false
	IstioEnabled bool `json:"istioEnabled,omitempty"`

	// VeleroEnabled describes if the cluster has a Velero running and enabled.
	// default: false
	VeleroEnabled bool `json:"veleroEnabled,omitempty"`

	// VeleroFsfreezeImage specifices the Velero Fsrfeeze image to use in Velero backup hooks
	// default: velero/fsfreeze-pause:latest
	VeleroFsfreezeImage string `json:"veleroFsfreezeImage,omitempty"`

	// InitContainers add extra initContainers
	VaultInitContainers []v1.Container `json:"vaultInitContainers,omitempty"`
}

// HAStorageTypes is the set of storage backends supporting High Availability
var HAStorageTypes = map[string]bool{
	"consul":     true,
	"dynamodb":   true,
	"etcd":       true,
	"gcs":        true,
	"mysql":      true,
	"postgresql": true,
	"raft":       true,
	"spanner":    true,
	"zookeeper":  true,
}

// HasHAStorage detects if Vault is configured to use a storage backend which supports High Availability or if it has
// ha_storage stanza, then doesn't check for ha_enabled flag
func (spec *VaultSpec) HasHAStorage() bool {
	storageType := spec.GetStorageType()
	if _, ok := HAStorageTypes[storageType]; ok {
		if spec.HasStorageHAEnabled() {
			return true
		}
	}
	if spec.hasHAStorageStanza() {
		return true
	}
	return false
}

func (spec *VaultSpec) hasHAStorageStanza() bool {
	return len(spec.getHAStorage()) != 0
}

// GetStorage returns Vault's storage stanza
func (spec *VaultSpec) GetStorage() map[string]interface{} {
	storage := spec.getStorage()
	return cast.ToStringMap(storage[spec.GetStorageType()])
}

func (spec *VaultSpec) getStorage() map[string]interface{} {
	config := spec.GetVaultConfig()
	return cast.ToStringMap(config["storage"])
}

// GetHAStorage returns Vault's ha_storage stanza
func (spec *VaultSpec) GetHAStorage() map[string]interface{} {
	haStorage := spec.getHAStorage()
	return cast.ToStringMap(haStorage[spec.GetHAStorageType()])
}

func (spec *VaultSpec) getHAStorage() map[string]interface{} {
	config := spec.GetVaultConfig()
	return cast.ToStringMap(config["ha_storage"])
}

func (spec *VaultSpec) GetVaultConfig() map[string]interface{} {
	var config map[string]interface{}
	// This config JSON is already validated,
	// so we can skip wiring through the error everywhere.
	_ = json.Unmarshal(spec.Config.Raw, &config)
	return config
}

// GetStorageType returns the type of Vault's storage stanza
func (spec *VaultSpec) GetStorageType() string {
	storage := spec.getStorage()
	return reflect.ValueOf(storage).MapKeys()[0].String()
}

// GetHAStorageType returns the type of Vault's ha_storage stanza
func (spec *VaultSpec) GetHAStorageType() string {
	haStorage := spec.getHAStorage()
	if len(haStorage) == 0 {
		return ""
	}
	return reflect.ValueOf(haStorage).MapKeys()[0].String()
}

// GetVersion returns the version of Vault
func (spec *VaultSpec) GetVersion() (*semver.Version, error) {
	version := strings.Split(spec.Image, ":")
	if len(version) != 2 {
		return nil, errors.New("failed to find Vault version")
	}
	return semver.NewVersion(version[1])
}

// GetServiceAccount returns the Kubernetes Service Account to use for Vault
func (spec *VaultSpec) GetServiceAccount() string {
	if spec.ServiceAccount != "" {
		return spec.ServiceAccount
	}
	return "default"
}

// HasStorageHAEnabled detects if the ha_enabled field is set to true in Vault's storage stanza
func (spec *VaultSpec) HasStorageHAEnabled() bool {
	storageType := spec.GetStorageType()
	storage := spec.getStorage()
	storageSpecs := cast.ToStringMap(storage[storageType])
	// In Consul HA is always enabled
	return storageType == "consul" || storageType == "raft" || cast.ToBool(storageSpecs["ha_enabled"])
}

// IsTLSDisabled returns if Vault's TLS should be disabled
func (spec *VaultSpec) IsTLSDisabled() bool {
	listener := spec.getListener()
	tcp := cast.ToStringMap(listener["tcp"])
	return cast.ToBool(tcp["tls_disable"])
}

// IsTelemetryUnauthenticated returns if Vault's telemetry endpoint can be accessed publicly
func (spec *VaultSpec) IsTelemetryUnauthenticated() bool {
	listener := spec.getListener()
	tcp := cast.ToStringMap(listener["tcp"])
	telemetry := cast.ToStringMap(tcp["telemetry"])
	return cast.ToBool(telemetry["unauthenticated_metrics_access"])
}

// GetAPIScheme returns if Vault's API address should be called on http or https
func (spec *VaultSpec) GetAPIScheme() string {
	if spec.IsTLSDisabled() {
		return "http"
	}
	return "https"
}

// GetTLSExpiryThreshold returns the Vault TLS certificate expiration threshold
func (spec *VaultSpec) GetTLSExpiryThreshold() time.Duration {
	if spec.TLSExpiryThreshold == "" {
		return time.Hour * 168
	}
	duration, err := time.ParseDuration(spec.TLSExpiryThreshold)
	if err != nil {
		log.Error(err, "using default threshold due to parse error", "tlsExpiryThreshold", spec.TLSExpiryThreshold)
		return time.Hour * 168
	}
	return duration
}

func (spec *VaultSpec) getListener() map[string]interface{} {
	config := spec.GetVaultConfig()
	return cast.ToStringMap(config["listener"])
}

// GetVaultImage returns the Vault image to use
func (spec *VaultSpec) GetVaultImage() string {
	if spec.Image == "" {
		return "vault:latest"
	}
	return spec.Image
}

// GetBankVaultsImage returns the bank-vaults image to use
func (spec *VaultSpec) GetBankVaultsImage() string {
	if spec.BankVaultsImage == "" {
		return bankVaultsImage
	}
	return spec.BankVaultsImage
}

// GetStatsDImage returns the StatsD image to use
func (spec *VaultSpec) GetStatsDImage() string {
	if spec.StatsDImage == "" {
		return "prom/statsd-exporter:latest"
	}
	return spec.StatsDImage
}

// GetVeleroFsfreezeImage returns the Velero Fsreeze image to use
func (spec *VaultSpec) GetVeleroFsfreezeImage() string {
	if spec.VeleroFsfreezeImage == "" {
		return "ubuntu:bionic"
	}
	return spec.VeleroFsfreezeImage
}

// GetVolumeClaimTemplates fixes the "status diff" in PVC templates
func (spec *VaultSpec) GetVolumeClaimTemplates() []v1.PersistentVolumeClaim {
	var pvcs []v1.PersistentVolumeClaim
	for _, pvc := range spec.VolumeClaimTemplates {
		pvcs = append(pvcs, v1.PersistentVolumeClaim{
			ObjectMeta: metav1.ObjectMeta{
				Name:        pvc.Name,
				Labels:      pvc.Labels,
				Annotations: pvc.Annotations,
			},
			Spec: pvc.Spec,
		})
	}
	return pvcs
}

// GetWatchedSecretsLabels returns the set of labels for secrets to watch in the vault namespace
func (spec *VaultSpec) GetWatchedSecretsLabels() []map[string]string {
	if spec.WatchedSecretsLabels == nil {
		spec.WatchedSecretsLabels = []map[string]string{}
	}

	return spec.WatchedSecretsLabels
}

// GetWatchedSecretsAnnotations returns the set of annotations for secrets to watch in the vault namespace
func (spec *VaultSpec) GetWatchedSecretsAnnotations() []map[string]string {
	if spec.WatchedSecretsAnnotations == nil {
		spec.WatchedSecretsAnnotations = []map[string]string{}
	}

	return spec.WatchedSecretsAnnotations
}

// GetAnnotations returns the Common Annotations
func (spec *VaultSpec) GetAnnotations() map[string]string {
	if spec.Annotations == nil {
		spec.Annotations = map[string]string{}
	}

	return spec.Annotations
}

// GetAPIPortName returns the main Vault port name based on Istio and TLS settings
func (spec *VaultSpec) GetAPIPortName() string {
	portName := "api-port"
	if spec.IstioEnabled {
		if spec.IsTLSDisabled() {
			return "http-" + portName
		}
		return "https-" + portName
	}
	return portName
}

// GetVaultLAbels returns the Vault Pod , Secret and ConfigMap Labels
func (spec *VaultSpec) GetVaultLabels() map[string]string {
	if spec.VaultLabels == nil {
		spec.VaultLabels = map[string]string{}
	}

	return spec.VaultLabels
}

// GetVaultConfigurerLabels returns the Vault Configurer Pod Labels
func (spec *VaultSpec) GetVaultConfigurerLabels() map[string]string {
	if spec.VaultConfigurerLabels == nil {
		spec.VaultConfigurerLabels = map[string]string{}
	}

	return spec.VaultConfigurerLabels
}

// GetVaultAnnotations returns the Vault Pod , Secret and ConfigMap Annotations
func (spec *VaultSpec) GetVaultAnnotations() map[string]string {
	if spec.VaultAnnotations == nil {
		spec.VaultAnnotations = map[string]string{}
	}

	return spec.VaultAnnotations
}

// GetVaultConfigurerAnnotations returns the Vault Configurer Pod Annotations
func (spec *VaultSpec) GetVaultConfigurerAnnotations() map[string]string {
	if spec.VaultConfigurerAnnotations == nil {
		spec.VaultConfigurerAnnotations = map[string]string{}
	}

	return spec.VaultConfigurerAnnotations
}

// GetFluentDImage returns the FluentD image to use
func (spec *VaultSpec) GetFluentDImage() string {
	if spec.FluentDImage == "" {
		return "fluent/fluentd:edge"
	}
	return spec.FluentDImage
}

// GetFluentDConfMountPath returns the mount path for the fluent.conf
func (spec *VaultSpec) GetFluentDConfMountPath() string {
	if spec.FluentDConfLocation == "" {
		return "/fluentd/etc"
	}
	return spec.FluentDConfLocation
}

// IsFluentDEnabled returns true if fluentd sidecar is to be deployed
func (spec *VaultSpec) IsFluentDEnabled() bool {
	return spec.FluentDEnabled
}

// IsStatsDDisabled returns false if statsd sidecar is to be deployed
func (spec *VaultSpec) IsStatsDDisabled() bool {
	return spec.StatsDDisabled
}

// ConfigJSON returns the Config field as a JSON string
func (v *Vault) ConfigJSON() ([]byte, error) {
	config := map[string]interface{}{}

	err := json.Unmarshal(v.Spec.Config.Raw, &config)
	if err != nil {
		return nil, err
	}

	if v.Spec.ServiceRegistrationEnabled && v.Spec.HasHAStorage() {
		serviceRegistration := map[string]interface{}{
			"service_registration": map[string]interface{}{
				"kubernetes": map[string]string{
					"namespace": v.Namespace,
				},
			},
		}

		if err := mergo.Merge(&config, serviceRegistration); err != nil {
			return nil, err
		}
	}

	configJSON, err := json.Marshal(config)
	if err != nil {
		return nil, err
	}

	return configJSON, nil
}

// ExternalConfigJSON returns the ExternalConfig field as a JSON string
func (spec *VaultSpec) ExternalConfigJSON() []byte {
	return spec.ExternalConfig.Raw
}

// IsAutoUnseal checks if auto-unseal is configured
func (spec *VaultSpec) IsAutoUnseal() bool {
	config := spec.GetVaultConfig()
	_, ok := config["seal"]
	return ok
}

// IsRaftStorage checks if raft storage is configured
func (spec *VaultSpec) IsRaftStorage() bool {
	return spec.GetStorageType() == "raft"
}

// IsRaftHAStorage checks if raft ha_storage is configured
func (spec *VaultSpec) IsRaftHAStorage() bool {
	return spec.GetStorageType() != "raft" && spec.GetHAStorageType() == "raft"
}

// IsRaftBootstrapFollower checks if this cluster should be considered the bootstrap follower.
func (spec *VaultSpec) IsRaftBootstrapFollower() bool {
	return spec.RaftLeaderAddress != "" && spec.RaftLeaderAddress != "self"
}

// GetIngress the Ingress configuration for Vault if any
func (vault *Vault) GetIngress() *Ingress {
	if vault.Spec.Ingress != nil {
		// Add the Vault Service as the backend if no rules are specified and there is no default backend
		if len(vault.Spec.Ingress.Spec.Rules) == 0 && vault.Spec.Ingress.Spec.DefaultBackend == nil {
			vault.Spec.Ingress.Spec.DefaultBackend = &netv1.IngressBackend{
				Service: &netv1.IngressServiceBackend{
					Name: vault.Name,
					Port: netv1.ServiceBackendPort{
						Number: 8200,
					},
				},
			}
		}

		if vault.Spec.Ingress.Annotations == nil {
			vault.Spec.Ingress.Annotations = map[string]string{}
		}

		// If TLS is enabled add the Ingress TLS backend annotations
		if !vault.Spec.IsTLSDisabled() {
			// Supporting the NGINX ingress controller with TLS backends
			// https://kubernetes.github.io/ingress-nginx/user-guide/nginx-configuration/annotations/#backend-protocol
			vault.Spec.Ingress.Annotations["nginx.ingress.kubernetes.io/backend-protocol"] = "HTTPS"

			// Supporting the Traefik ingress controller with TLS backends
			// https://docs.traefik.io/configuration/backends/kubernetes/#tls-communication-between-traefik-and-backend-pods
			vault.Spec.Ingress.Annotations["ingress.kubernetes.io/protocol"] = "https"

			// Supporting the HAProxy ingress controller with TLS backends
			// https://github.com/jcmoraisjr/haproxy-ingress#secure-backend
			vault.Spec.Ingress.Annotations["ingress.kubernetes.io/secure-backends"] = "true"
		}

		return vault.Spec.Ingress
	}

	return nil
}

// LabelsForVault returns the labels for selecting the resources
// belonging to the given vault CR name.
func (vault *Vault) LabelsForVault() map[string]string {
	return map[string]string{"app.kubernetes.io/name": "vault", "vault_cr": vault.Name}
}

// LabelsForVaultConfigurer returns the labels for selecting the resources
// belonging to the given vault CR name.
func (vault *Vault) LabelsForVaultConfigurer() map[string]string {
	return map[string]string{"app.kubernetes.io/name": "vault-configurator", "vault_cr": vault.Name}
}

// AsOwnerReference returns this Vault instance as an OwnerReference
func (vault *Vault) AsOwnerReference() metav1.OwnerReference {
	return metav1.OwnerReference{
		APIVersion: vault.APIVersion,
		Kind:       vault.Kind,
		Name:       vault.Name,
		UID:        vault.UID,
		Controller: pointer.BoolPtr(true),
	}
}

// VaultStatus defines the observed state of Vault
type VaultStatus struct {
	// Important: Run "make generate-code" to regenerate code after modifying this file
	Nodes      []string                `json:"nodes"`
	Leader     string                  `json:"leader"`
	Conditions []v1.ComponentCondition `json:"conditions,omitempty"`
}

// UnsealConfig represents the UnsealConfig field of a VaultSpec Kubernetes object
type UnsealConfig struct {
	Options    UnsealOptions          `json:"options,omitempty"`
	Kubernetes KubernetesUnsealConfig `json:"kubernetes,omitempty"`
	Google     *GoogleUnsealConfig    `json:"google,omitempty"`
	Alibaba    *AlibabaUnsealConfig   `json:"alibaba,omitempty"`
	Azure      *AzureUnsealConfig     `json:"azure,omitempty"`
	AWS        *AWSUnsealConfig       `json:"aws,omitempty"`
	Vault      *VaultUnsealConfig     `json:"vault,omitempty"`
	HSM        *HSMUnsealConfig       `json:"hsm,omitempty"`
}

// UnsealOptions represents the common options to all unsealing backends
type UnsealOptions struct {
	PreFlightChecks *bool `json:"preFlightChecks,omitempty"`
	StoreRootToken  *bool `json:"storeRootToken,omitempty"`
}

func (uso UnsealOptions) ToArgs() []string {
	args := []string{}
	if uso.PreFlightChecks == nil || *uso.PreFlightChecks {
		args = append(args, "--pre-flight-checks=true")
	}
	if uso.StoreRootToken != nil && !*uso.StoreRootToken {
		args = append(args, "--store-root-token=false")
	}

	return args
}

// ToArgs returns the UnsealConfig as and argument array for bank-vaults
func (usc *UnsealConfig) ToArgs(vault *Vault) []string {
	args := []string{}

	if usc.Options.PreFlightChecks == nil || *usc.Options.PreFlightChecks {
		args = append(args,
			"--pre-flight-checks=true")
	}

	if usc.Options.StoreRootToken != nil && !*usc.Options.StoreRootToken {
		args = append(args,
			"--store-root-token=false")
	}

	if usc.Google != nil {

		args = append(args,
			"--mode",
			"google-cloud-kms-gcs",
			"--google-cloud-kms-key-ring",
			usc.Google.KMSKeyRing,
			"--google-cloud-kms-crypto-key",
			usc.Google.KMSCryptoKey,
			"--google-cloud-kms-location",
			usc.Google.KMSLocation,
			"--google-cloud-kms-project",
			usc.Google.KMSProject,
			"--google-cloud-storage-bucket",
			usc.Google.StorageBucket,
		)

	} else if usc.Azure != nil {

		args = append(args,
			"--mode",
			"azure-key-vault",
			"--azure-key-vault-name",
			usc.Azure.KeyVaultName,
		)

	} else if usc.AWS != nil {

		args = append(args,
			"--mode",
			"aws-kms-s3",
			"--aws-kms-key-id",
			usc.AWS.KMSKeyID,
			"--aws-kms-region",
			usc.AWS.KMSRegion,
			"--aws-s3-bucket",
			usc.AWS.S3Bucket,
			"--aws-s3-prefix",
			usc.AWS.S3Prefix,
			"--aws-s3-region",
			usc.AWS.S3Region,
			"--aws-s3-sse-algo",
			usc.AWS.S3SSE,
		)

	} else if usc.Alibaba != nil {

		args = append(args,
			"--mode",
			"alibaba-kms-oss",
			"--alibaba-kms-region",
			usc.Alibaba.KMSRegion,
			"--alibaba-kms-key-id",
			usc.Alibaba.KMSKeyID,
			"--alibaba-oss-endpoint",
			usc.Alibaba.OSSEndpoint,
			"--alibaba-oss-bucket",
			usc.Alibaba.OSSBucket,
			"--alibaba-oss-prefix",
			usc.Alibaba.OSSPrefix,
		)

	} else if usc.Vault != nil {

		args = append(args,
			"--mode",
			"vault",
			"--vault-addr",
			usc.Vault.Address,
			"--vault-unseal-keys-path",
			usc.Vault.UnsealKeysPath,
		)

		if usc.Vault.Token != "" {
			args = append(args,
				"--vault-token",
				usc.Vault.Token,
			)
		} else if usc.Vault.TokenPath != "" {
			args = append(args,
				"--vault-token-path",
				usc.Vault.TokenPath,
			)
		} else if usc.Vault.Role != "" {
			args = append(args,
				"--vault-role",
				usc.Vault.Role,
				"--vault-auth-path",
				usc.Vault.AuthPath,
			)
		}

	} else if usc.HSM != nil {

		mode := "hsm"
		if usc.Kubernetes.SecretNamespace != "" && usc.Kubernetes.SecretName != "" {
			mode = "hsm-k8s"
		}

		args = append(args,
			"--mode",
			mode,
			"--hsm-module-path",
			usc.HSM.ModulePath,
			"--hsm-slot-id",
			fmt.Sprint(usc.HSM.SlotID),
			"--hsm-key-label",
			usc.HSM.KeyLabel,
			"--hsm-pin",
			usc.HSM.Pin,
		)

		if usc.HSM.TokenLabel != "" {
			args = append(args,
				"--hsm-token-label",
				usc.HSM.TokenLabel,
			)
		}

		if mode == "hsm-k8s" {
			var secretLabels []string
			for k, v := range vault.LabelsForVault() {
				secretLabels = append(secretLabels, k+"="+v)
			}

			sort.Strings(secretLabels)

			args = append(args,
				"--k8s-secret-namespace",
				usc.Kubernetes.SecretNamespace,
				"--k8s-secret-name",
				usc.Kubernetes.SecretName,
				"--k8s-secret-labels",
				strings.Join(secretLabels, ","),
			)
		}

	} else {

		secretNamespace := vault.Namespace
		if usc.Kubernetes.SecretNamespace != "" {
			secretNamespace = usc.Kubernetes.SecretNamespace
		}

		secretName := vault.Name + "-unseal-keys"
		if usc.Kubernetes.SecretName != "" {
			secretName = usc.Kubernetes.SecretName
		}

		var secretLabels []string
		for k, v := range vault.LabelsForVault() {
			secretLabels = append(secretLabels, k+"="+v)
		}

		sort.Strings(secretLabels)

		args = append(args,
			"--mode",
			"k8s",
			"--k8s-secret-namespace",
			secretNamespace,
			"--k8s-secret-name",
			secretName,
			"--k8s-secret-labels",
			strings.Join(secretLabels, ","),
		)

	}

	return args
}

// HSMDaemonNeeded returns if the unsealing mechanims needs a HSM Daemon present
func (usc *UnsealConfig) HSMDaemonNeeded() bool {
	return usc.HSM != nil && usc.HSM.Daemon
}

// KubernetesUnsealConfig holds the parameters for Kubernetes based unsealing
type KubernetesUnsealConfig struct {
	SecretNamespace string `json:"secretNamespace,omitempty"`
	SecretName      string `json:"secretName,omitempty"`
}

// GoogleUnsealConfig holds the parameters for Google KMS based unsealing
type GoogleUnsealConfig struct {
	KMSKeyRing    string `json:"kmsKeyRing"`
	KMSCryptoKey  string `json:"kmsCryptoKey"`
	KMSLocation   string `json:"kmsLocation"`
	KMSProject    string `json:"kmsProject"`
	StorageBucket string `json:"storageBucket"`
}

// AlibabaUnsealConfig holds the parameters for Alibaba Cloud KMS based unsealing
//  --alibaba-kms-region eu-central-1 --alibaba-kms-key-id 9d8063eb-f9dc-421b-be80-15d195c9f148 --alibaba-oss-endpoint oss-eu-central-1.aliyuncs.com --alibaba-oss-bucket bank-vaults
type AlibabaUnsealConfig struct {
	KMSRegion   string `json:"kmsRegion"`
	KMSKeyID    string `json:"kmsKeyId"`
	OSSEndpoint string `json:"ossEndpoint"`
	OSSBucket   string `json:"ossBucket"`
	OSSPrefix   string `json:"ossPrefix"`
}

// AzureUnsealConfig holds the parameters for Azure Key Vault based unsealing
type AzureUnsealConfig struct {
	KeyVaultName string `json:"keyVaultName"`
}

// AWSUnsealConfig holds the parameters for AWS KMS based unsealing
type AWSUnsealConfig struct {
	KMSKeyID  string `json:"kmsKeyId"`
	KMSRegion string `json:"kmsRegion,omitempty"`
	S3Bucket  string `json:"s3Bucket"`
	S3Prefix  string `json:"s3Prefix"`
	S3Region  string `json:"s3Region,omitempty"`
	S3SSE     string `json:"s3SSE,omitempty"`
}

// VaultUnsealConfig holds the parameters for remote Vault based unsealing
type VaultUnsealConfig struct {
	Address        string `json:"address"`
	UnsealKeysPath string `json:"unsealKeysPath"`
	Role           string `json:"role,omitempty"`
	AuthPath       string `json:"authPath,omitempty"`
	TokenPath      string `json:"tokenPath,omitempty"`
	Token          string `json:"token,omitempty"`
}

// HSMUnsealConfig holds the parameters for remote HSM based unsealing
type HSMUnsealConfig struct {
	Daemon     bool   `json:"daemon,omitempty"`
	ModulePath string `json:"modulePath"`
	SlotID     uint   `json:"slotId,omitempty"`
	TokenLabel string `json:"tokenLabel,omitempty"`
	Pin        string `json:"pin"`
	KeyLabel   string `json:"keyLabel"`
}

// CredentialsConfig configuration for a credentials file provided as a secret
type CredentialsConfig struct {
	Env        string `json:"env"`
	Path       string `json:"path"`
	SecretName string `json:"secretName"`
}

// Resources holds different container's ResourceRequirements
type Resources struct {
	Vault              *v1.ResourceRequirements `json:"vault,omitempty"`
	BankVaults         *v1.ResourceRequirements `json:"bankVaults,omitempty"`
	HSMDaemon          *v1.ResourceRequirements `json:"hsmDaemon,omitempty"`
	PrometheusExporter *v1.ResourceRequirements `json:"prometheusExporter,omitempty"`
	FluentD            *v1.ResourceRequirements `json:"fluentd,omitempty"`
}

// Ingress specification for the Vault cluster
type Ingress struct {
	Annotations map[string]string `json:"annotations,omitempty"`
	Spec        netv1.IngressSpec `json:"spec,omitempty"`
}


text:
using System;
using System.Collections.Specialized;
using System.Configuration;

namespace Aliencube.ConfigurationEncryptor.ConsoleApp
{
    /// <summary>
    /// This represents the cryption service entity.
    /// </summary>
    internal class CryptionService
    {
        private readonly Parameter _param;
        private readonly NameValueCollection _settings;

        /// <summary>
        /// Initialises a new instance of the CryptionService class.
        /// </summary>
        /// <param name="param">Parameter instance.</param>
        /// <param name="settings">AppSettings instance.</param>
        public CryptionService(Parameter param, NameValueCollection settings)
        {
            this._param = param;
            this._settings = settings;
        }

        private string _encryptionProvider;

        /// <summary>
        /// Gets the encryption provider.
        /// </summary>
        public string EncryptionProvider
        {
            get
            {
                if (String.IsNullOrWhiteSpace(this._encryptionProvider))
                    this._encryptionProvider = this._settings["EncryptionProvider"];

                return this._encryptionProvider;
            }
        }

        /// <summary>
        /// Executes encryption or decryption.
        /// </summary>
        /// <returns>Returns the configuration path encrypted/decrypted.</returns>
        public string Execute()
        {
            string filepath;
            switch (this._param.Direction)
            {
                case Direction.Encrypt:
                    filepath = this.EncryptSections(this._param.Filename, this._param.Sections);
                    break;

                case Direction.Decrypt:
                    filepath = this.DecryptSections(this._param.Filename, this._param.Sections);
                    break;

                default:
                    throw new InvalidOperationException("Must set the direction - Encrypt or Decrypt");
            }

            return filepath;
        }

        /// <summary>
        /// Encrypts configuration sections.
        /// </summary>
        /// <param name="filename">App.config/Web.config filename.</param>
        /// <param name="sections">Sections enum value.</param>
        /// <returns>Returns the configuration file path.</returns>
        private string EncryptSections(string filename, Sections sections)
        {
            var configuration = ConfigurationManager.OpenExeConfiguration(filename);
            if (sections.HasFlag(Sections.ConnectionStrings))
            {
                var connectionStringsSection = configuration.GetSection("connectionStrings") as ConnectionStringsSection;
                if (IsSectionAvailable(connectionStringsSection, Direction.Encrypt))
                {
                    connectionStringsSection.SectionInformation.ProtectSection(this.EncryptionProvider);
                    connectionStringsSection.SectionInformation.ForceSave = true;
                }
            }

            if (sections.HasFlag(Sections.AppSettings))
            {
                var appSettingsSection = configuration.GetSection("appSettings") as AppSettingsSection;
                if (IsSectionAvailable(appSettingsSection, Direction.Encrypt))
                {
                    appSettingsSection.SectionInformation.ProtectSection(this.EncryptionProvider);
                    appSettingsSection.SectionInformation.ForceSave = true;
                }
            }
            configuration.Save();
            return configuration.FilePath;
        }

        /// <summary>
        /// Decrypts configuration sections.
        /// </summary>
        /// <param name="filename">App.config/Web.config filename.</param>
        /// <param name="sections">Sections enum value.</param>
        /// <returns>Returns the configuration file path.</returns>
        private string DecryptSections(string filename, Sections sections)
        {
            var configuration = ConfigurationManager.OpenExeConfiguration(filename);
            if (sections.HasFlag(Sections.ConnectionStrings))
            {
                var connectionStringsSection = configuration.GetSection("connectionStrings") as ConnectionStringsSection;
                if (IsSectionAvailable(connectionStringsSection, Direction.Decrypt))
                {
                    connectionStringsSection.SectionInformation.UnprotectSection();
                    connectionStringsSection.SectionInformation.ForceSave = true;
                }
            }

            if (sections.HasFlag(Sections.AppSettings))
            {
                var appSettingsSection = configuration.GetSection("appSettings") as AppSettingsSection;
                if (IsSectionAvailable(appSettingsSection, Direction.Decrypt))
                {
                    appSettingsSection.SectionInformation.UnprotectSection();
                    appSettingsSection.SectionInformation.ForceSave = true;
                }
            }
            configuration.Save();
            return configuration.FilePath;
        }

        /// <summary>
        /// Checks whether the configuration section is available or not.
        /// </summary>
        /// <param name="section">ConfigurationSection instance.</param>
        /// <param name="direction">Direction to encrypt or decrypt.</param>
        /// <returns>Returns <c>True</c>, if the configuration section is available; otherwise returns <c>False</c>.</returns>
        private bool IsSectionAvailable(ConfigurationSection section, Direction direction)
        {
            if (section == null || section.ElementInformation.IsLocked || section.SectionInformation.IsLocked)
                return false;

            if (direction == Direction.Encrypt)
                return !section.SectionInformation.IsProtected;

            if (direction == Direction.Decrypt)
                return section.SectionInformation.IsProtected;

            return false;
        }
    }
}

redpajama_set_name: RedPajamaGithub


text:
package org.hammerlab.guacamole.commands

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.hammerlab.guacamole.{ DistributedUtil, SparkCommand, Common }
import org.hammerlab.guacamole.reads.{ PairedMappedRead, Read }
import org.kohsuke.args4j.{ Option => Args4JOption }

import scalax.collection.Graph
import scalax.collection.edge._
import scalax.collection.mutable.{ Graph => MutableGraph }
import scalax.collection.GraphPredef._
import scalax.collection.edge.Implicits._

/**
 * Structural Variant caller
 *
 * This follows the DELLY algorithm:
 * 1. Look for read pairs with abnormally large insert sizes.
 * 2. Build a graph of pairs which could be explained by a single SV.
 * 3. Find maximal cliques in this graph.
 *
 * These cliques each represent a large deletion.
 *
 * The output is a text file of genomic ranges where there's some evidence of an SV.
 */
object StructuralVariant {

  val MAX_INSERT_SIZE = 25000
  val BLOCK_SIZE = 25

  protected class Arguments extends Common.Arguments.Reads with DistributedUtil.Arguments {
    @Args4JOption(name = "--filter-contig", usage = "Filter to alignments where either mate is in this contig.")
    var filterContig: String = ""

    @Args4JOption(name = "--output", usage = "Path to output text file.")
    var output: String = ""
  }

  case class Location(
    contig: String,
    position: Long)

  case class GenomeRange(
    contig: String,
    start: Long,
    stop: Long)

  case class MedianStats(
    median: Double,
    mad: Double)

  // An undirected, weighted graph of evidence for structural variants.
  // Nodes are paired reads.
  // Edges represent compatibility between the induced structural variants,
  // with the weight being the strength of compatibility.
  type PairedReadGraph = Graph[PairedMappedRead, WkUnDiEdge]

  object Caller extends SparkCommand[Arguments] {
    override val name = "structural-variant"
    override val description = "Find structural variants, e.g. large deletions."

    // The sign of inferredInsertSize follows the orientation of the read. This returns
    // an insert size which is positive in the common case, regardless of the orientation.
    def orientedInsertSize(r: PairedMappedRead): Int = {
      val sgn = if (r.read.isPositiveStrand) { 1 } else { -1 }
      r.inferredInsertSize * sgn
    }

    // Extract median and MAD (Median Absolute Deviation) from an unordered Array.
    def medianStats[T <% Double](xs: Array[T])(implicit ord: Ordering[T]): MedianStats = {

      // Extract the median of a sorted, non-empty array of numbers
      def getMedian[T <% Double](nums: Array[T]): Double = {
        val len = nums.length
        if (len % 2 == 0) {
          0.5 * (nums(len / 2 - 1) + nums(len / 2))
        } else {
          1.0 * nums(len / 2)
        }
      }

      if (xs.isEmpty) {
        MedianStats(0.0, 0.0)
      } else {
        val nums = xs.sorted
        val median = getMedian(nums)
        val residuals = nums.map(x => Math.abs(1.0 * x - median)).sorted
        val mad = getMedian(residuals)

        MedianStats(median, mad)
      }
    }

    case class ExceptionalReadsReturnType(
      readsInRange: RDD[PairedMappedRead],
      insertSizes: RDD[Int],
      insertStats: MedianStats,
      maxNormalInsertSize: Int,
      exceptionalReads: RDD[PairedMappedRead])

    def getExceptionalReads(reads: RDD[PairedMappedRead]): ExceptionalReadsReturnType = {
      // Pare down to mate pairs which aren't highly displaced & aren't inverted or translocated.
      val readsInRange = reads.filter(r => {
        val s = r.inferredInsertSize
        r.read.referenceContig == r.mate.referenceContig &&
          r.read.isPositiveStrand != r.mate.isPositiveStrand &&
          s < MAX_INSERT_SIZE
      })
      readsInRange.persist()

      val insertSizes = readsInRange.map(orientedInsertSize)
      val insertStats = medianStats(insertSizes.takeSample(false, 100000))
      println("insert stats: " + insertStats)

      // Find read pairs where the insert size differs significantly from the genome-wide median.
      // Following DELLY, we use a threshold of 5 median absolute deviations above the median.
      val maxNormalInsertSize = (insertStats.median + 5 * insertStats.mad).toInt

      val exceptionalReads = readsInRange.filter(_.inferredInsertSize > maxNormalInsertSize)

      ExceptionalReadsReturnType(
        readsInRange,
        insertSizes,
        insertStats,
        maxNormalInsertSize,
        exceptionalReads
      )
    }

    // Given two pairs of reads, is there a deletion which would make their insert sizes normal?
    def areReadsCompatible(read1: PairedMappedRead,
                           read2: PairedMappedRead,
                           maxNormalInsertSize: Long): Boolean = {
      if (read1.minPos > read2.minPos) {
        areReadsCompatible(read2, read1, maxNormalInsertSize)
      } else {
        // This matches the DELLY logic; the last five lines are copy/paste. See
        // https://github.com/tobiasrausch/delly/blob/7464cacfe1d420ac5bd7ed40a3093a80a93a6964/src/tags.h#L441-L445
        val (pair1Min, pair1GapMin, pair1GapMax, pair1Max) = read1.startsAndStops
        val pair1ReadLength = read1.readLength

        val (pair2Min, pair2GapMin, pair2GapMax, pair2Max) = read2.startsAndStops
        val pair2ReadLength = read2.readLength

        !(((pair2GapMin - pair1Min) > maxNormalInsertSize) ||
          ((pair2GapMax < pair1GapMax) && ((pair1Max - pair2GapMax) > maxNormalInsertSize)) ||
          ((pair2GapMax >= pair1GapMax) && ((pair2Max - pair1GapMax) > maxNormalInsertSize)) ||
          ((pair1GapMax < pair2Min) || (pair2GapMax < pair1Min)))
      }
    }

    // Given a list of reads with abnormally large inserts, construct a graph representation of the
    // structural variants where:
    // Vertices = paired reads
    // Edges = two sets of paired reads could represent the same structural variant
    // Weights = compatibility between reads (0=most compatible)
    // For more details, see the DELLY paper.
    def buildVariantGraph(exceptionalReads: Iterable[PairedMappedRead],
                          maxNormalInsertSize: Int): PairedReadGraph = {
      val reads = exceptionalReads.toArray.sortBy(_.minPos)
      val graph = MutableGraph[PairedMappedRead, WkUnDiEdge]()

      for { (read, i) <- reads.zipWithIndex } {
        val (start, _, gapEnd, _) = read.startsAndStops
        var j = i + 1
        var done = false
        // Loop through subsequent reads until it's impossible to find another compatible one.
        while (j < reads.length && !done) {
          val nextRead = reads(j)
          val (nextStart, nextGapStart, nextGapEnd, _) = nextRead.startsAndStops
          if (Math.abs(nextGapStart - start) > maxNormalInsertSize) {
            done = true
          } else {
            if (areReadsCompatible(read, nextRead, maxNormalInsertSize)) {
              // TODO: this is from DELLY but seems imprecise. It only considers the insert size,
              // not position. If two reads are of similar length, but don't overlap much, they'll
              // get a spuriously low weight.
              val weight = Math.abs((nextGapEnd - nextStart) - (gapEnd - start))
              graph += WkUnDiEdge(read, nextRead)(weight)
            }
          }
          j += 1
        }
      }

      // TODO: exclude isolated reads from the graph
      graph
    }

    case class SVClique(readPairs: Set[PairedMappedRead],
                        wiggle: Long,
                        svStart: Long,
                        svEnd: Long,
                        maxNormalInsertSize: Int) {

      // All connected components will have at least one node (by definition), so .head is safe.
      def span = GenomeRange(readPairs.head.read.referenceContig, svStart, svEnd)

      // If an edge has one node in the clique and one not, return the one that's not.
      def findEligibleNode(edge: PairedReadGraph#EdgeT): Option[PairedReadGraph#NodeT] = {
        val Seq(node1, node2) = edge.nodes.toSeq
        val node1InClique = readPairs.contains(node1.value)
        val node2InClique = readPairs.contains(node2.value)

        // The edge should have one end in the clique and one end out of it.
        (node1InClique, node2InClique) match {
          case (false, true) => Some(node1)
          case (true, false) => Some(node2)
          case _             => None
        }
      }

      def nodeIsConnected(node: PairedReadGraph#NodeT): Boolean = {
        readPairs.subsetOf(node.neighbors map { _.value })
      }

      def maybeAbsorbNode(n: PairedReadGraph#NodeT): Option[SVClique] = {
        val (_, nodeGapMin, nodeGapMax, _) = n.value.startsAndStops
        val newSvStart = Math.max(svStart, nodeGapMin)
        val newSvEnd = Math.min(svEnd, nodeGapMax)

        // The amount of wiggle room could shrink either because of the new read, or because it
        // shrinks the induced SV and pulls another read closer to being incompatible.
        val wiggleNewRead = maxNormalInsertSize - (n.value.insertSize - (newSvEnd - newSvStart))
        val wiggleChange = wiggle + (newSvEnd - newSvStart) - (svEnd - svStart)
        val newWiggle = Math.min(wiggleNewRead, wiggleChange)

        if ((newSvStart < newSvEnd) && (newWiggle >= 0)) {
          Some(SVClique(readPairs + n.value, newWiggle, newSvStart, newSvEnd, maxNormalInsertSize))
        } else {
          None
        }
      }
    }

    object SVClique {
      def apply(pair: PairedMappedRead, maxNormalInsertSize: Int): SVClique = {
        val (_, svStart, svEnd, _) = pair.startsAndStops
        // this is the "wiggle room": by how much can the SV shrink and still be compatible with all
        // reads in the clique? It starts positive and then shrinks towards zero (at which point
        // there's no wiggle room left).
        val wiggle = maxNormalInsertSize - ((pair.insertSize - (svEnd - svStart)))
        SVClique(Set(pair), wiggle, svStart, svEnd, maxNormalInsertSize)
      }
    }

    def findOneClique(g: PairedReadGraph, maxNormalInsertSize: Int): SVClique = {
      // Seed the clique with the source of the lowest-weight edge.
      // Because the graph is constructed by adding edges, all connected components will have at
      // least one edge.
      val edges = g.edges.toList.sortBy(_.weight)
      val bestEdge = edges.head

      val initPair = bestEdge.nodes.minBy(_.minPos).value

      edges.foldLeft(SVClique(initPair, maxNormalInsertSize))((clique, edge) => {
        clique
          .findEligibleNode(edge)
          .filter(clique.nodeIsConnected)
          .flatMap(clique.maybeAbsorbNode)
          .getOrElse(clique)
      })
    }

    def findCliques(g: PairedReadGraph, maxNormalInsertSize: Int): Traversable[SVClique] = {
      for {
        c <- g.componentTraverser()
        if c.nodes.size >= 2
      } yield findOneClique(c.toGraph, maxNormalInsertSize)
    }

    override def run(args: Arguments, sc: SparkContext): Unit = {
      val readSet = Common.loadReadsFromArguments(args, sc, Read.InputFilters(nonDuplicate = true))
      val pairedMappedReads = readSet.mappedPairedReads
      val firstInPair = pairedMappedReads.filter(_.isFirstInPair).flatMap(PairedMappedRead(_))

      val ExceptionalReadsReturnType(_, _, _, maxNormalInsertSize, exceptionalReads) =
        getExceptionalReads(firstInPair)

      val readsByContig = exceptionalReads.groupBy(_.read.referenceContig)
      val svGraph = readsByContig.mapValues(buildVariantGraph(_, maxNormalInsertSize))
      val svs = svGraph.mapValues(findCliques(_, maxNormalInsertSize))
        .mapValues(_.map(_.span).toList)

      svs.coalesce(1).saveAsTextFile(args.output)
    }
  }
}


text:
<?xml version="1.0" encoding="UTF-8"?>
<settings xmlns="http://maven.apache.org/SETTINGS/1.0.0"
          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
          xsi:schemaLocation="http://maven.apache.org/SETTINGS/1.0.0 http://maven.apache.org/xsd/settings-1.0.0.xsd">

    <servers>
        <server>
            <id>bintray-sedovalx-com.github.sedovalx</id>
            <username>${BINTRAY_USERNAME}</username>
            <password>${BINTRAY_API_KEY}</password>
        </server>
    </servers>

</settings>


text:
package org.openkilda.floodlight.service.kafka;

import static org.easymock.EasyMock.anyObject;
import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.getCurrentArguments;
import static org.easymock.EasyMock.replay;
import static org.easymock.EasyMock.verify;

import org.openkilda.messaging.info.InfoMessage;
import org.openkilda.messaging.info.event.PortChangeType;
import org.openkilda.messaging.info.event.PortInfoData;
import org.openkilda.model.SwitchId;

import net.floodlightcontroller.core.module.FloodlightModuleContext;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.easymock.Capture;
import org.easymock.EasyMock;
import org.easymock.EasyMockSupport;
import org.easymock.IAnswer;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class KafkaProducerServiceTest extends EasyMockSupport {
    private static final String TOPIC = "A";
    private static final TopicPartition[] partitions = new TopicPartition[]{
            new TopicPartition(TOPIC, 0),
            new TopicPartition(TOPIC, 1)
    };

    private KafkaProducerService subject;

    @SuppressWarnings("unchecked")
    private Producer<String, String> kafkaProducer = (Producer<String, String>) strictMock(Producer.class);

    @Before
    public void setUp() {
        injectMocks(this);

        FloodlightModuleContext moduleContext = new FloodlightModuleContext();

        KafkaUtilityService kafkaUtility = createMock(KafkaUtilityService.class);
        expect(kafkaUtility.makeProducer()).andReturn(kafkaProducer);
        moduleContext.addService(KafkaUtilityService.class, kafkaUtility);

        replay(kafkaUtility);

        subject = new KafkaProducerService();
        subject.setup(moduleContext);

        ArrayList<PartitionInfo> partitionsForResult = new ArrayList<>(2);
        for (TopicPartition p : partitions) {
            partitionsForResult.add(new PartitionInfo(p.topic(), p.partition(), null, null, null));
        }
        expect(kafkaProducer.partitionsFor(TOPIC)).andReturn(partitionsForResult).anyTimes();
    }

    @Test
    public void errorReporting() throws Exception {
        final ExecutionException error = new ExecutionException("Emulate kafka send error", new IOException());

        Future promise = mock(Future.class);
        expect(promise.get()).andThrow(error).anyTimes();
        replay(promise);
        expect(kafkaProducer.send(anyObject(), anyObject(Callback.class)))
                .andAnswer(new IAnswer<Future<RecordMetadata>>() {
                    @Override
                    public Future<RecordMetadata> answer() {
                        Callback callback = (Callback) getCurrentArguments()[1];
                        callback.onCompletion(null, error);
                        return promise;
                    }
                });

        replay(kafkaProducer);
        subject.sendMessageAndTrack(TOPIC, makePayload());
        verify(kafkaProducer);

        // This test does not do any assertions, because the only action is log message with error
        // you can locate this message in test's output.
    }

    @Test
    public void errorDetection() throws Exception {
        Future promise = mock(Future.class);
        final ExecutionException error = new ExecutionException("Emulate kafka send error", new IOException());
        expect(promise.get()).andThrow(error).anyTimes();
        replay(promise);

        expect(kafkaProducer.send(anyObject(), anyObject(Callback.class))).andReturn(promise);

        replay(kafkaProducer);
        SendStatus status = subject.sendMessage(TOPIC, makePayload());
        verify(kafkaProducer);

        Boolean isThrown;
        try {
            status.waitTillComplete();
            isThrown = false;
        } catch (ExecutionException e) {
            isThrown = true;
        }
        Assert.assertTrue(String.format(
                "Exception was not thrown by %s object", status.getClass().getCanonicalName()), isThrown);
    }

    private InfoMessage makePayload() {
        return new InfoMessage(
                new PortInfoData(new SwitchId("ff:fe:00:00:00:00:00:01"), 8, PortChangeType.UP),
                System.currentTimeMillis(), getClass().getCanonicalName() + "-test");
    }

    @SuppressWarnings("unchecked")
    private void setupSendCapture(Capture<ProducerRecord<String, String>> trap, RecordMetadata[] sendResults)
            throws Exception {
        for (RecordMetadata metadata : sendResults) {
            Future promise = mock(Future.class);
            expect(promise.get()).andReturn(metadata);
            replay(promise);

            expect(kafkaProducer.send(EasyMock.capture(trap), anyObject(Callback.class)))
                    .andReturn(promise);
        }
    }
}


text:
package omap

import "strings"

// NewStringKeyed returns an empty Map that accepts case-sensitive
// string keys.
func NewStringKeyed() *Map {
    return &Map{less: func(a, b interface{}) bool {
        return a.(string) < b.(string)
    }}
}

// NewCaseFoldedKeyed returns an empty Map that accepts case-insensitive
// string keys.
func NewCaseFoldedKeyed() *Map {
    return &Map{less: func(a, b interface{}) bool {
        return strings.ToLower(a.(string)) < strings.ToLower(b.(string))
    }}
}

// NewIntKeyed returns an empty Map that accepts int keys.
func NewIntKeyed() *Map {
    return &Map{less: func(a, b interface{}) bool {
        return a.(int) < b.(int)
    }}
}

// NewFloat64Keyed returns an empty Map that accepts float64 keys.
func NewFloat64Keyed() *Map {
    return &Map{less: func(a, b interface{}) bool {
        return a.(float64) < b.(float64)
    }}
}

// New returns an empty Map that uses the given less than function to
// compare keys. For example:
//      type Point { X, Y int }
//      pointMap := omap.New(func(a, b interface{}) bool {
//              α, β := a.(Point), b.(Point)
//              if α.X != β.X {
//                  return α.X < β.X
//              }
//              return α.Y < β.Y
//          })
func New(less func(interface{}, interface{}) bool) *Map {
    return &Map{less: less}
}

// Map is a key-ordered map.
// The zero value is an invalid map! Use one of the construction functions
// (e.g., New()), to create a map for a specific key type.
type Map struct {
    root   *node
    less   func(interface{}, interface{}) bool
    length int
}

type node struct {
    key, value  interface{}
    red         bool
    left, right *node
}

// Insert inserts a new key-value into the Map and returns true; or
// replaces an existing key-value pair's value if the keys are equal and
// returns false. For example:
//      inserted := myMap.Insert(key, value).
func (m *Map) Insert(key, value interface{}) (inserted bool) {
    m.root, inserted = m.insert(m.root, key, value)
    m.root.red = false
    if inserted {
        m.length++
    }
    return inserted
}

// Find returns the value and true if the key is in the Map or nil and
// false otherwise. For example:
//      value, found := myMap.Find(key).
func (m *Map) Find(key interface{}) (value interface{}, found bool) {
    root := m.root
    for root != nil {
        if m.less(key, root.key) {
            root = root.left
        } else if m.less(root.key, key) {
            root = root.right
        } else {
            return root.value, true
        }
    }
    return nil, false
}

// Delete deletes the key-value with the given key from the Map and returns
// true, or does nothing and returns false if there is no key-value with
// the given key. For example:
//      deleted := myMap.Delete(key).
func (m *Map) Delete(key interface{}) (deleted bool) {
    if m.root != nil {
        if m.root, deleted = m.remove(m.root, key); m.root != nil {
            m.root.red = false
        }
    }
    if deleted {
        m.length--
    }
    return deleted
}

// Do calls the given function on every key-value in the Map in order.
func (m *Map) Do(function func(interface{}, interface{})) {
    do(m.root, function)
}

// Len returns the number of key-value pairs in the map.
func (m *Map) Len() int {
    return m.length
}

func (m *Map) insert(root *node, key, value interface{}) (*node, bool) {
    inserted := false
    if root == nil { // If the key was in the tree it would belong here
        return &node{key: key, value: value, red: true}, true
    }
    if isRed(root.left) && isRed(root.right) {
        colorFlip(root)
    }
    if m.less(key, root.key) {
        root.left, inserted = m.insert(root.left, key, value)
    } else if m.less(root.key, key) {
        root.right, inserted = m.insert(root.right, key, value)
    } else { // The key is already in the tree so just replace its value
        root.value = value
    }
    if isRed(root.right) && !isRed(root.left) {
        root = rotateLeft(root)
    }
    if isRed(root.left) && isRed(root.left.left) {
        root = rotateRight(root)
    }
    return root, inserted
}

func isRed(root *node) bool { return root != nil && root.red }

func colorFlip(root *node) {
    root.red = !root.red
    if root.left != nil {
        root.left.red = !root.left.red
    }
    if root.right != nil {
        root.right.red = !root.right.red
    }
}

func rotateLeft(root *node) *node {
    x := root.right
    root.right = x.left
    x.left = root
    x.red = root.red
    root.red = true
    return x
}

func rotateRight(root *node) *node {
    x := root.left
    root.left = x.right
    x.right = root
    x.red = root.red
    root.red = true
    return x
}

func do(root *node, function func(interface{}, interface{})) {
    if root != nil {
        do(root.left, function)
        function(root.key, root.value)
        do(root.right, function)
    }
}

// We do not provide an exported First() method because this is an
// implementation detail.
func first(root *node) *node {
    for root.left != nil {
        root = root.left
    }
    return root
}

func (m *Map) remove(root *node, key interface{}) (*node, bool) {
    deleted := false
    if m.less(key, root.key) {
        if root.left != nil {
            if !isRed(root.left) && !isRed(root.left.left) {
                root = moveRedLeft(root)
            }
            root.left, deleted = m.remove(root.left, key)
        }
    } else {
        if isRed(root.left) {
            root = rotateRight(root)
        }
        if !m.less(key, root.key) && !m.less(root.key, key) &&
            root.right == nil {
            return nil, true
        }
        if root.right != nil {
            if !isRed(root.right) && !isRed(root.right.left) {
                root = moveRedRight(root)
            }
            if !m.less(key, root.key) && !m.less(root.key, key) {
                smallest := first(root.right)
                root.key = smallest.key
                root.value = smallest.value
                root.right = deleteMinimum(root.right)
                deleted = true
            } else {
                root.right, deleted = m.remove(root.right, key)
            }
        }
    }
    return fixUp(root), deleted
}

func moveRedLeft(root *node) *node {
    colorFlip(root)
    if root.right != nil && isRed(root.right.left) {
        root.right = rotateRight(root.right)
        root = rotateLeft(root)
        colorFlip(root)
    }
    return root
}

func moveRedRight(root *node) *node {
    colorFlip(root)
    if root.left != nil && isRed(root.left.left) {
        root = rotateRight(root)
        colorFlip(root)
    }
    return root
}

func deleteMinimum(root *node) *node {
    if root.left == nil {
        return nil
    }
    if !isRed(root.left) && !isRed(root.left.left) {
        root = moveRedLeft(root)
    }
    root.left = deleteMinimum(root.left)
    return fixUp(root)
}

func fixUp(root *node) *node {
    if isRed(root.right) {
        root = rotateLeft(root)
    }
    if isRed(root.left) && isRed(root.left.left) {
        root = rotateRight(root)
    }
    if isRed(root.left) && isRed(root.right) {
        colorFlip(root)
    }
    return root
}


text:
package com.hazelcast.jet.impl.connector;

import com.hazelcast.collection.IList;
import com.hazelcast.jet.JetInstance;
import com.hazelcast.jet.Util;
import com.hazelcast.jet.core.DAG;
import com.hazelcast.jet.core.JetTestSupport;
import com.hazelcast.jet.core.Vertex;
import com.hazelcast.test.HazelcastParallelClassRunner;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.util.Collections;
import java.util.Map.Entry;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.Future;
import java.util.concurrent.locks.LockSupport;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import static com.hazelcast.jet.Util.entry;
import static com.hazelcast.jet.core.Edge.between;
import static com.hazelcast.jet.core.processor.SinkProcessors.writeListP;
import static com.hazelcast.jet.core.processor.SourceProcessors.streamFilesP;
import static com.hazelcast.jet.impl.util.ExceptionUtil.sneakyThrow;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

@RunWith(HazelcastParallelClassRunner.class)
@Ignore
public class StreamFilesP_integrationTest extends JetTestSupport {

    private JetInstance instance;
    private File directory;
    private IList<Entry<String, String>> list;

    @Before
    public void setup() throws Exception {
        instance = createJetMember();
        directory = createTempDirectory();
        list = instance.getList("writer");
    }

    @Test
    public void when_appendingToPreexisting_then_pickupNewLines() throws Exception {
        DAG dag = buildDag();

        // this is a pre-existing file, should not be picked up
        File file = createNewFile();
        appendToFile(file, "hello", "pre-existing");
        sleepAtLeastMillis(50);

        Future<Void> jobFuture = instance.newJob(dag).getFuture();
        // wait for the processor to initialize
        sleepAtLeastSeconds(2);

        // pre-existing file should not be picked up
        assertEquals(0, list.size());
        appendToFile(file, "third line");
        // now, all three lines are picked up
        assertTrueEventually(() -> assertEquals(1, list.size()));

        finishDirectory(jobFuture, file);
    }

    @Test
    public void when_appendingToPreexistingIncompleteLine_then_pickupCompleteLines() throws Exception {
        DAG dag = buildDag();

        // this is a pre-existing file, should not be picked up
        File file = createNewFile();
        try (PrintWriter writer = new PrintWriter(new FileOutputStream(file, true))) {
            // note: no newline appended
            writer.write("hello");
        }
        sleepAtLeastMillis(50);

        Future<Void> jobFuture = instance.newJob(dag).getFuture();
        // wait for the processor to initialize
        sleepAtLeastSeconds(2);

        // pre-existing file should not be picked up
        assertEquals(0, list.size());
        // this completes the first line and a second one - only the second one should be picked
        appendToFile(file, "world", "second line");
        // now, all three lines are picked up
        assertTrueEventually(() -> assertEquals(1, list.size()));
        assertEquals(entry(file.getName(), "second line"), list.get(0));

        finishDirectory(jobFuture, file);
    }

    @Test
    public void when_withCrlf_then_pickupCompleteLines() throws Exception {
        DAG dag = buildDag();

        // this is a pre-existing file, should not be picked up
        File file = createNewFile();
        try (PrintWriter writer = new PrintWriter(new FileOutputStream(file, true))) {
            // note: no newline appended
            writer.write("hello world\r");
        }
        sleepAtLeastMillis(50);

        Future<Void> jobFuture = instance.newJob(dag).getFuture();
        // wait for the processor to initialize
        sleepAtLeastSeconds(2);

        // pre-existing file should not be picked up
        assertEquals(0, list.size());
        // this completes the first line and a second one - only the second one should be picked
        try (PrintWriter writer = new PrintWriter(new FileOutputStream(file, true))) {
            // note: no newline appended
            writer.write("\nsecond line\r\n");
        }
        // now, all three lines are picked up
        assertTrueEventually(() -> assertEquals(1, list.size()));
        assertEquals(entry(file.getName(), "second line"), list.get(0));

        finishDirectory(jobFuture, file);
    }

    @Test
    @Ignore
    public void when_newAndModified_then_pickupAddition() throws Exception {
        DAG dag = buildDag();

        Future<Void> jobFuture = instance.newJob(dag).getFuture();
        // wait for the processor to initialize
        sleepAtLeastSeconds(2);

        assertEquals(0, list.size());
        File file = new File(directory, randomName());
        appendToFile(file, "hello", "world");
        assertTrueEventually(() -> assertEquals(2, list.size()));

        // now add one more line, only this line should be picked up
        appendToFile(file, "third line");
        assertTrueEventually(() -> assertEquals(3, list.size()));

        finishDirectory(jobFuture, file);
    }

    @Test
    public void when_fileWithManyLines_then_emitCooperatively() throws Exception {
        DAG dag = buildDag();

        Future<Void> jobFuture = instance.newJob(dag).getFuture();
        // wait for the processor to initialize
        sleepAtLeastSeconds(2);

        assertEquals(0, list.size());
        File subdir = new File(directory, "subdir");
        assertTrue(subdir.mkdir());
        File fileInSubdir = new File(subdir, randomName());
        appendToFile(fileInSubdir, IntStream.range(0, 5000).mapToObj(String::valueOf).toArray(String[]::new));

        // move the file to watchedDirectory
        File file = new File(directory, fileInSubdir.getName());
        assertTrue(fileInSubdir.renameTo(file));

        assertTrueEventually(() -> assertEquals(5000, list.size()));

        finishDirectory(jobFuture, file, subdir);
    }

    @Test
    @Ignore
    public void stressTest() throws Exception {
        // Here, I will start the job and in two separate threads I'll write fixed number of lines
        // to two different log files, with few ms hiccups between each line.
        // At the end, I'll check, if all the contents matches.
        DAG dag = buildDag();

        Future<Void> jobFuture = instance.newJob(dag).getFuture();
        // wait for the processor to initialize
        sleepAtLeastSeconds(2);
        int numLines = 10_000;
        File file1 = new File(directory, "file1.log");
        File file2 = new File(directory, "file2.log");
        Thread t1 = new Thread(new RandomWriter(file1, numLines, "file0"));
        Thread t2 = new Thread(new RandomWriter(file2, numLines, "file1"));
        t1.start();
        t2.start();
        t1.join();
        t2.join();

        // wait for the list to be full
        assertTrueEventually(() -> assertEquals(2 * numLines, list.size()));

        // check the list
        Set<Integer>[] expectedNumbers = new Set[] {
                IntStream.range(0, numLines).boxed().collect(Collectors.toSet()),
                IntStream.range(0, numLines).boxed().collect(Collectors.toSet())};

        for (Entry<String, String> logLine : list) {
            // logLine has the form "fileN M ...", N is fileIndex, M is line number
            int fileIndex = logLine.getValue().charAt(4) - '0';
            int lineIndex = Integer.parseInt(logLine.getValue().split(" ", 3)[1]);
            assertTrue(expectedNumbers[fileIndex].remove(lineIndex));
        }

        assertEquals(Collections.emptySet(), expectedNumbers[0]);
        assertEquals(Collections.emptySet(), expectedNumbers[1]);

        finishDirectory(jobFuture, file1, file2);
    }

    private static class RandomWriter implements Runnable {

        private final File file;
        private final int numLines;
        private final String prefix;
        private final Random random = new Random();

        RandomWriter(File file, int numLines, String prefix) {
            this.file = file;
            this.numLines = numLines;
            this.prefix = prefix;
        }

        @Override
        public void run() {
            try (FileOutputStream fos = new FileOutputStream(file);
                    OutputStreamWriter osw = new OutputStreamWriter(fos, UTF_8);
                    BufferedWriter bw = new BufferedWriter(osw)
            ) {
                for (int i = 0; i < numLines; i++) {
                    bw.write(prefix + ' ' + i +
                            " Lorem ipsum dolor sit amet, consectetur adipiscing elit," +
                            " sed do eiusmod tempor incididunt ut labore et dolore magna aliqua\n");
                    bw.flush();
                    osw.flush();
                    fos.flush();
                    if (random.nextInt(100) < 5) {
                        LockSupport.parkNanos(MILLISECONDS.toNanos(1));
                    }
                }
            } catch (IOException e) {
                throw sneakyThrow(e);
            }
        }
    }

    private DAG buildDag() {
        DAG dag = new DAG();
        Vertex reader = dag.newVertex("reader", streamFilesP(directory.getPath(), UTF_8, "*", false, Util::entry))
                           .localParallelism(1);
        Vertex writer = dag.newVertex("writer", writeListP(list.getName())).localParallelism(1);
        dag.edge(between(reader, writer));
        return dag;
    }

    private File createNewFile() {
        File file = new File(directory, randomName());
        assertTrueEventually(() -> assertTrue(file.createNewFile()));
        return file;
    }

    private void finishDirectory(Future<Void> jobFuture, File ... files) throws Exception {
        for (File file : files) {
            logger.info("deleting " + file + "...");
            assertTrueEventually(() -> assertTrue("Failed to delete " + file, file.delete()));
            logger.info("deleted " + file);
        }
        assertTrueEventually(() -> assertTrue("Failed to delete " + directory, directory.delete()));
        assertTrueEventually(() -> assertTrue("job should complete eventually", jobFuture.isDone()));
        // called for side-effect of throwing exception, if the job failed
        jobFuture.get();
    }
}


text:
class IdentifierScheme < ApplicationRecord
  include FlagShihTzu
  include Subscribable

  ##
  # The maximum length for a name
  NAME_MAXIMUM_LENGTH = 30

  has_many :identifiers

  # ===============
  # = Validations =
  # ===============

  validates :name, uniqueness: { message: UNIQUENESS_MESSAGE },
                   presence: { message: PRESENCE_MESSAGE },
                   length: { maximum: NAME_MAXIMUM_LENGTH }

  validates :active, inclusion: { message: INCLUSION_MESSAGE,
                                  in: BOOLEAN_VALUES }

  # ===========================
  # = Scopes =
  # ===========================

  scope :active, -> { where(active: true) }
  scope :by_name, ->(value) { active.where('LOWER(name) = LOWER(?)', value) }

  ##
  # Define Bit Field values for the scheme's context
  # These are used to determine when and where an identifier scheme is applicable
  #   for_authentication => identifies which schemes can be used for user auth
  #   for_orgs           => identifies which ids will be displayed on Org pages
  #   for_plans          => identifies which ids will be displayed on Plans pages
  #   for_contributors   => identifies which ids will be displayed on Contributor pages
  #   for_identification => identifies which ids are object identifiers (e.g. ROR, ARK, etc.)
  has_flags 1 => :for_authentication,
            2 => :for_orgs,
            3 => :for_plans,
            4 => :for_users,
            5 => :for_contributors,
            6 => :for_identification,
            7 => :for_research_outputs,
            column: 'context'

  # =========================
  # = Custom Accessor Logic =
  # =========================

  # The name is used by the OrgSearchService as a key. For example:
  #    { "ror": "12345" }
  # so we cannot allow spaces or non alpha characters!
  def name=(value)
    super(value&.downcase&.gsub(/[^a-z]/, ''))
  end

  # ===========================
  # = Instance Methods =
  # ===========================
end

# -----------------------------------------------------
# Bitwise key
# -----------------------------------------------------
# 01 - for_authentication
# 02 - for_orgs
# 03 - for_authentication + for_orgs
# 04 - for_plans
# 05 - for_authentication + for_plans
# 06 - for_orgs + for_plans
# 07 - for_authentication + for_plans + for_orgs
# 08 - for_users
# 09 - for_authentication + for_users
# 10 - for_orgs + for_users
# 11 - for_authentication + for_orgs + for_users
# 12 - for_plans + for_users
# 13 - for_authentication + for_plans + for_users
# 14 - for_orgs + for_plans + for_users
# 15 - for_authentication + for_orgs + for_plans + for_users
# 16 - for_contributors
# 17 - for_authentication + for_contributors
# 18 - for_orgs + for_contributors
# 19 - for_authentication + for_orgs + for_contributors
# 20 - for_plans + for_contributors
# 21 - for_authentication + for_plans + for_contributors
# 22 - administraor + for_plans + for_contributors
# 23 - for_authentication + for_plans + for_orgs + for_contributors
# 24 - for_users + for_contributors
# 25 - for_authentication + for_users + for_contributors
# 26 - for_orgs + for_users + for_contributors
# 27 - for_authentication + for_orgs + for_users + for_contributors
# 28 - for_plans + for_users + for_contributors
# 29 - for_authentication + for_plans + for_users + for_contributors
# 30 - for_orgs + for_plans + for_users + for_contributors
# 31 - for_authentication + for_orgs + for_plans + for_users + for_contributors
# ... (32 combos)
# 63 - for_identification
# ... (64 combos)
# 127 - for_research_outputs
# ... (128 combos)


text:
package uk.gov.hmrc.apigateway.connector

import play.api.libs.json.Format
import play.api.libs.ws.WSClient
import uk.gov.hmrc.apigateway.cache.CacheManager
import uk.gov.hmrc.apigateway.util.PlayRequestUtils
import uk.gov.hmrc.play.config.ServicesConfig

import scala.concurrent.Future
import scala.reflect.ClassTag

abstract class ServiceConnector(wsClient: WSClient, cache: CacheManager, val serviceName: String)
  extends AbstractConnector(wsClient) with ServicesConfig {

  lazy val serviceBaseUrl = baseUrl(serviceName)

  override def get[T: ClassTag](urlPath: String)(implicit format: Format[T]): Future[T] =
    get(urlPath, urlPath, Seq.empty)

  def get[T: ClassTag](key: String, urlPath: String)(implicit format: Format[T]): Future[T] =
    get(key, urlPath, Seq.empty)

  def get[T: ClassTag](key: String, urlPath: String, headers: Seq[(String, String)])(implicit format: Format[T]): Future[T] =
    cache.get[T](key, serviceName, super.get(s"$serviceBaseUrl/$urlPath", headers), PlayRequestUtils.asMapOfSets(headers))
}


text:
package org.atoum.intellij.plugin.atoum.run;

import com.intellij.execution.ExecutionException;
import com.intellij.execution.Executor;
import com.intellij.execution.configurations.ConfigurationFactory;
import com.intellij.execution.configurations.GeneralCommandLine;
import com.intellij.execution.configurations.RunConfiguration;
import com.intellij.execution.executors.DefaultRunExecutor;
import com.intellij.execution.process.*;
import com.intellij.execution.testframework.TestConsoleProperties;
import com.intellij.execution.testframework.TestFrameworkRunningModel;
import com.intellij.execution.testframework.sm.SMTestRunnerConnectionUtil;
import com.intellij.execution.testframework.sm.runner.SMTRunnerConsoleProperties;
import com.intellij.execution.testframework.sm.runner.SMTestProxy;
import com.intellij.execution.testframework.sm.runner.ui.SMTRunnerConsoleView;
import com.intellij.execution.testframework.sm.runner.ui.TestResultsViewer;
import com.intellij.execution.testframework.ui.BaseTestsOutputConsoleView;
import com.intellij.notification.Notification;
import com.intellij.notification.NotificationType;
import com.intellij.notification.Notifications;
import com.intellij.openapi.project.Project;
import com.intellij.openapi.util.Disposer;
import com.intellij.openapi.util.Key;
import com.intellij.openapi.vfs.VirtualFile;
import com.intellij.openapi.wm.ToolWindow;
import com.intellij.openapi.wm.ToolWindowAnchor;
import com.intellij.openapi.wm.ToolWindowManager;
import com.intellij.psi.PsiDirectory;
import com.intellij.ui.content.Content;
import com.intellij.ui.content.ContentManager;
import com.jetbrains.php.config.PhpProjectConfigurationFacade;
import com.jetbrains.php.config.interpreters.PhpConfigurationOptionData;
import com.jetbrains.php.config.interpreters.PhpInterpreter;
import com.jetbrains.php.run.PhpRunConfiguration;
import com.jetbrains.php.run.PhpRunConfigurationFactoryBase;
import org.atoum.intellij.plugin.atoum.AtoumUtils;
import org.atoum.intellij.plugin.atoum.model.*;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.atoum.intellij.plugin.atoum.Icons;
import java.io.File;

import java.util.*;

public class Runner {

    protected Project project;

    public Runner(Project project)
    {
        this.project = project;
    }

    public void run(RunnerConfiguration runnerConfiguration)
    {
        ToolWindow toolWindow = getToolWindow();
        getConsole(toolWindow, project, runnerConfiguration);
    }

    public ToolWindow getToolWindow() {
        ToolWindow toolWindow;

        String windowId = "atoum";
        if (!Arrays.asList(ToolWindowManager.getInstance(project).getToolWindowIds()).contains(windowId)) {
            toolWindow = ToolWindowManager.getInstance(project).registerToolWindow(windowId, false, ToolWindowAnchor.BOTTOM);
        } else {
            toolWindow = ToolWindowManager.getInstance(project).getToolWindow(windowId);
        }

        toolWindow.setIcon(Icons.ATOUM);
        toolWindow.show(null);

        return toolWindow;
    }

    // like
    //   https://github.com/JetBrains/intellij-community/blob/6555e34dc0b5c4bd3a8c9efc7d8e5ca84929af40/platform/platform-impl/src/com/intellij/execution/process/ScriptRunnerUtil.java
    // with not log and charset support
    // but with environment variables support
    protected OSProcessHandler prepareProcessHandler(@NotNull String exePath, @Nullable String workingDirectory, String[] parameters, @Nullable Map<String, String> environment) throws ExecutionException {
        GeneralCommandLine commandLine = new GeneralCommandLine(exePath);
        commandLine.addParameters(parameters);
        if (workingDirectory != null) {
            commandLine.setWorkDirectory(workingDirectory);
        }

        commandLine.withEnvironment(environment);

        return new ColoredProcessHandler(commandLine);
    }

    public BaseTestsOutputConsoleView getConsole(ToolWindow toolWindow, Project project, final RunnerConfiguration runnerConfiguration) {

        ConfigurationFactory myFactory = new PhpRunConfigurationFactoryBase(new AtoumLocalRunConfigurationType()) {
            public RunConfiguration createTemplateConfiguration(Project project) {
                return new AtoumLocalRunConfiguration(project, this, "");
            }
        };

        Executor executor = DefaultRunExecutor.getRunExecutorInstance();
        PhpRunConfiguration runConfiguration = new AtoumLocalRunConfiguration(project, myFactory, "test");
        TestConsoleProperties testConsoleProperties = new SMTRunnerConsoleProperties(runConfiguration, "atoum", executor);
        final BaseTestsOutputConsoleView testsOutputConsoleView = SMTestRunnerConnectionUtil.createConsole("atoumConsole", testConsoleProperties);

        Disposer.register(project, testsOutputConsoleView);

        VirtualFile testBaseDir = null;
        try {
            testBaseDir = AtoumUtils.findTestBaseDir(runnerConfiguration, project);
        } catch (Exception e) {
            testBaseDir = project.getBaseDir();
        }

        String testBasePath = testBaseDir.getPath();
        String atoumBinPath = AtoumUtils.findAtoumBinPath(testBaseDir);

        String phpPath = "php";
        PhpInterpreter interpreter = PhpProjectConfigurationFacade.getInstance(project).getInterpreter();
        if (null != interpreter) {
            phpPath = interpreter.getPathToPhpExecutable();
        }

        List<PhpConfigurationOptionData> phpConfig;
        try {
            phpConfig = interpreter.getConfigurationOptions();
        } catch (NullPointerException e) {
            phpConfig = new ArrayList<PhpConfigurationOptionData>();
        }

        CommandLineArgumentsBuilder commandLineBuilder = (new CommandLineArgumentsBuilder(atoumBinPath, testBasePath, phpConfig))
            .useTapReport()
            .useConfiguration(runnerConfiguration)
        ;

        String phpstormConfigFile = testBasePath + "/.atoum.phpstorm.php";
        if (new File(phpstormConfigFile).exists()) {
            commandLineBuilder.useConfigFile(phpstormConfigFile);
        }

        ContentManager contentManager = toolWindow.getContentManager();
        Content myContent;
        myContent = toolWindow.getContentManager().getFactory().createContent(testsOutputConsoleView.getComponent(), "tests results", false);
        toolWindow.getContentManager().removeAllContents(true);
        toolWindow.getContentManager().addContent(myContent);

        final SMTRunnerConsoleView console = (SMTRunnerConsoleView)testsOutputConsoleView;

        String[] commandLineArgs = commandLineBuilder.build();

        HashMap environnmentVariables = new HashMap();
        environnmentVariables.put("PHPSTORM", "1");

        OSProcessHandler processHandler = null;
        try {
            processHandler = this.prepareProcessHandler(
                phpPath,
                testBasePath,
                commandLineArgs,
                environnmentVariables
            );

            testsOutputConsoleView.attachToProcess(processHandler);
            console.getResultsViewer().setAutoscrolls(true);
            TestConsoleProperties.HIDE_PASSED_TESTS.set(testsOutputConsoleView.getProperties(), true);
            TestConsoleProperties.SELECT_FIRST_DEFECT.set(testsOutputConsoleView.getProperties(), true);

            final OSProcessHandler finalProcessHandler = processHandler;
            console.getResultsViewer().addEventsListener(new TestResultsViewer.EventsListener() {

                final StringBuilder outputBuilder = new StringBuilder();

                @Override
                public void onTestingStarted(TestResultsViewer testResultsViewer) {

                    if (finalProcessHandler != null) {
                        finalProcessHandler.addProcessListener(new ProcessAdapter() {
                            public void onTextAvailable(ProcessEvent event, Key outputType) {
                                outputBuilder.append(event.getText());
                            }
                            public void processTerminated(ProcessEvent event) {
                            }
                        });
                    }
                }

                @Override
                public void onTestingFinished(TestResultsViewer testResultsViewer) {
                    SMTestProxy testsRootNode = testResultsViewer.getTestsRootNode();

                    if (outputBuilder.length() == 0) {
                        testsRootNode.setTestFailed("No tests were found!", "", true);
                        return;
                    }

                    TestsResult testsResult = TestsResultFactory.createFromTapOutput(outputBuilder.toString());

                    SMTRootTestProxyFactory.updateFromTestResult(testsResult, testsRootNode);

                    selectFirstFailedMethod();

                    if (testsResult.getState().equals(testsResult.STATE_PASSED)) {
                        TestConsoleProperties.HIDE_PASSED_TESTS.set(testsOutputConsoleView.getProperties(), false);
                    }
                }

                protected void selectFirstFailedMethod()
                {
                    for (SMTestProxy testProxy: console.getResultsViewer().getTestsRootNode().getAllTests()) {
                        for (SMTestProxy methodProxy: testProxy.getAllTests()) {
                            if (methodProxy.isDefect()) {
                                console.getResultsViewer().selectAndNotify(methodProxy);
                            }
                        }

                    }
                }

                @Override
                public void onTestNodeAdded(TestResultsViewer testResultsViewer, SMTestProxy smTestProxy) {

                }

                @Override
                public void onSelected(@Nullable SMTestProxy smTestProxy, @NotNull TestResultsViewer testResultsViewer, @NotNull TestFrameworkRunningModel testFrameworkRunningModel) {

                }
            });

            processHandler.startNotify();

        } catch (ExecutionException e) {
            Notifications.Bus.notify(new Notification("atoumGroup", "Error running tests", e.getMessage(), NotificationType.ERROR, null), project);
        }

        return testsOutputConsoleView;
    }
}


text:
from . import trigger

class pretrigger(trigger):
    pre = True


text:
package org.dozer.vo.direction;

/**
 * @author dmitry.buzdin
 */
public class ContentItemGroupDefault extends ContentItemGroupBase implements ContentItemGroup {

}


text:
<?xml version="1.0" encoding="utf-8"?>
<FrameLayout xmlns:android="http://schemas.android.com/apk/res/android"
    android:layout_width="match_parent"
    android:layout_height="match_parent" >

    <ProgressBar
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:layout_gravity="center"
        android:animationResolution="33"
        android:indeterminateDrawable="@drawable/progress_medium_holo" />

    <com.inveno.piflow.entity.view.news.NewsViewPager
        android:id="@+id/meituviewpager"
        android:layout_width="match_parent"
        android:layout_height="match_parent" >
    </com.inveno.piflow.entity.view.news.NewsViewPager>

</FrameLayout>


text:
import { Component, OnInit } from '@angular/core';
import { FormsModule } from '@angular/forms';
import { UserService } from './user.service';
import { UserLoginClass } from './user-login-class';

@Component({
    selector: 'user-login',
    template: `

Hello LOGIN form

                  Error:
                  {{message}}

                 <form #userForm='ngForm' >

                Name
                <input type="text" class="form-control" [(ngModel)]='currentUser.name' id="name"  required name='name' #nameControl='ngModel'>

                    Name field is required

                Password
                <input type="password" class="form-control" id="pass" [(ngModel)]='currentUser.password' required name='pass' #passControl='ngModel'>

                    Password field is required

             <button type="submit" [disabled]='userForm.invalid' class="btn btn-default" (click)='login(nameControl, passControl)' >new user</button>

        </form>

You submitted the following:

Name

{{ currentUser.name }}

password

{{ currentUser.password }}

Sity

{{ currentUser.city }}

            <button type="submit"  class="btn btn-default" (click)='logOut()' >Log out</button>

    `,
    styles: [
        '.ng-valid[required], .ng-valid.required{border-left:solid green 5px}',
        '.ng-invalid:not(form){border-left: solid red 5px}',
        '.div-with-message{height:80px}'

    ]
})
export class UserLoginComponent implements OnInit {
    test = true;
    currentUser = new UserLoginClass('', '');
    registred: boolean = false;
    registredUser: any;
    message: string = '';
    classForMessage: {};

    constructor(
        private userService: UserService,

    ) { }

    ngOnInit() {
        this.checkInLocalStorageUser();
    }

    checkInLocalStorageUser() {
        let storageUser = this.userService.checkInLocalStorageUser();
        if (storageUser) {
            this.currentUser = new UserLoginClass(storageUser.name, storageUser.password);
            this.registred = true;
        }
        else {
            this.currentUser = new UserLoginClass('', '')
            this.registred = false
        }
        console.log(this.currentUser)
    }

    login() {
        this.userService.login(this.currentUser)
            .subscribe(data => {
                if (data) {
                    console.log(data)
                    window.localStorage.setItem('currentUser', JSON.stringify({ name: data['name'], password: data['password'], city: data['city'] }));
                    this.registred = true;
                    this.showMessage(`user:${this.currentUser.name} successfully found, ${this.registred}`)

                } else {
                    console.log('not User');
                    this.registred = false;
                    this.showMessage(`user not found, ${this.registred}`)
                }
            })
    }

    logOut() {
        this.currentUser = new UserLoginClass('', '');
        window.localStorage.removeItem('currentUser');
        this.userService.setLogged(false)
        this.registred = false;
        window.localStorage['currentUser'] ?
            console.log('current user:' + window.localStorage['currentUser']) :
            console.log('there is not current user');
    }

    showMessage(msg: string) {
        this.message = msg;
        this.setCurrentClasses();
        setTimeout(() => this.message = '', 2000)

    }

    setCurrentClasses() {
        this.classForMessage = {
            'alert': true,
            'alert-success': this.registred,
            'alert-danger': !this.registred

        }
    }
}


text:
<?php
defined('C5_EXECUTE') or die("Access Denied.");
class FileSetPermissionResponse extends Concrete5_Model_FileSetPermissionResponse {}


text:
echo "Build postgres-addon-arcgis docker image ..."
cd docker
docker build -t huas/postgres-addon-arcgis:9.5 .
docker tag huas/postgres-addon-arcgis:9.5 \
    huas/postgres-addon-arcgis:latest
cd ..
echo "remove postgres-addon-arcgis docker images ..."
docker rmi $(docker images -q -f dangling=true)
docker rmi huas/postgres-addon-arcgis:9.5


text:
<?php

namespace Being\CacheModel;

class Model extends \Angejia\Pea\Model
{
}