text:
using namespace Urho3D;

//=============================================================================
//=============================================================================
const int CTRL_FORWARD = 1;
const int CTRL_BACK = 2;
const int CTRL_LEFT = 4;
const int CTRL_RIGHT = 8;
const int CTRL_JUMP = 16;

const float MOVE_FORCE = 0.8f;
const float INAIR_MOVE_FORCE = 0.06f;
const float BRAKE_FORCE = 0.2f;
const float JUMP_FORCE = 7.0f;
const float YAW_SENSITIVITY = 0.01f;
const float INAIR_THRESHOLD_TIME = 0.1f;

//** collision layers and masks**
enum ColLayerType
{
    ColLayer_None       = (0),
    ColLayer_Static     = (1<<0),
    ColLayer_Character  = (1<<1),
    ColLayer_World      = (1<<2),
};

enum ColMaskType
{
    ColMask_All          = ~(ColLayer_None),
    ColMask_Static       = ColMask_All,
    ColMask_Character    = ColMask_All,
    ColMask_World        = ColMask_All,
    ColMask_WorldRayCast = ~(ColLayer_Static | ColLayer_Character),
};

//=============================================================================
//=============================================================================
/// Character component, responsible for physical movement according to controls, as well as animation.
class Character : public LogicComponent
{
    URHO3D_OBJECT(Character, LogicComponent);

public:
    /// Construct.
    Character(Context* context);

    /// Register object factory and attributes.
    static void RegisterObject(Context* context);

    /// Handle startup. Called by LogicComponent base class.
    virtual void Start();
    /// Handle physics world update. Called by LogicComponent base class.
    virtual void FixedUpdate(float timeStep);
    virtual void FixedPostUpdate(float timeStep);

public:
    Controls controls_;

protected:
    /// Handle physics collision event.
    void HandleNodeCollision(StringHash eventType, VariantMap& eventData);
    void UpdateOrientation();

protected:
    /// Grounded flag for movement.
    bool onGround_;
    /// Jump flag.
    bool okToJump_;
    /// In air timer. Due to possible physics inaccuracy, character can be off ground for max. 1/10 second and still be allowed to move.
    float inAirTimer_;

    // extra vars
    Vector3 curMoveDir_;
    bool isJumping_;
    bool jumpStarted_;

    // world sphere node
    WeakPtr<Node>  worldSphereNode_;
    Vector3        dirToCenterWorld_;
    float          prevYaw_;
};


text:
require 'mail'

module Banking
  # Class for handling everything email-related
  class Email < Grape::API
    format :json

    Mail.defaults do
      delivery_method :smtp, address: 'mail.example.com',
                             domain: 'example.com',
                             user_name: 'user',
                             password: 'password',
                             authentication: :plain,
                             openssl_verify_mode: 'none'
    end

    post :register do
      mail = Mail.new
      mail.to = params[:email]
      mail.from = 'Banking Test <bankingtest@example.com>'
      mail.subject = 'More info about ' << params[:account]
      mail.body = File.open('info/' << params[:account]) { |file| file.read }
      mail.deliver!
      { success: true }
    end
  end
end

redpajama_set_name: RedPajamaGithub


text:
@implementation TEColorCrossfade

+(UIColor*)colorForFadeBetweenFirstColor:(UIColor *)firstColor secondColor:(UIColor *)secondColor atRatio:(CGFloat)ratio {

    if (CGColorGetColorSpace(firstColor.CGColor) != CGColorGetColorSpace(secondColor.CGColor))
    {
        firstColor = [TEColorCrossfade colorConvertedToRGBA:firstColor];
        secondColor = [TEColorCrossfade colorConvertedToRGBA:secondColor];
    }

    // Grab color components
    const CGFloat *firstColorComponents = CGColorGetComponents(firstColor.CGColor);
    const CGFloat *secondColorComponents = CGColorGetComponents(secondColor.CGColor);

    // Interpolate between colors
    CGFloat interpolatedComponents[CGColorGetNumberOfComponents(firstColor.CGColor)] ;
    for (NSUInteger i = 0; i < CGColorGetNumberOfComponents(firstColor.CGColor); i++)
    {
        interpolatedComponents[i] = firstColorComponents[i] * (1 - ratio) + secondColorComponents[i] * ratio;
    }

    // Create interpolated color
    CGColorRef interpolatedCGColor = CGColorCreate(CGColorGetColorSpace(firstColor.CGColor), interpolatedComponents);
    UIColor *interpolatedColor = [UIColor colorWithCGColor:interpolatedCGColor];
    CGColorRelease(interpolatedCGColor);

    return interpolatedColor;
}

+(UIColor *)colorConvertedToRGBA:(UIColor *)colorToConvert {
    CGFloat red;
    CGFloat green;
    CGFloat blue;
    CGFloat alpha;

    // Convert color to RGBA with a CGContext. UIColor's getRed:green:blue:alpha: doesn't work across color spaces. Adapted from http://stackoverflow.com/a/4700259

    alpha = CGColorGetAlpha(colorToConvert.CGColor);

    CGColorRef opaqueColor = CGColorCreateCopyWithAlpha(colorToConvert.CGColor, 1.0f);
    CGColorSpaceRef rgbColorSpace = CGColorSpaceCreateDeviceRGB();
    unsigned char resultingPixel[CGColorSpaceGetNumberOfComponents(rgbColorSpace)];
    CGContextRef context = CGBitmapContextCreate(&resultingPixel, 1, 1, 8, 4, rgbColorSpace, (CGBitmapInfo)kCGImageAlphaNoneSkipLast);
    CGContextSetFillColorWithColor(context, opaqueColor);
    CGColorRelease(opaqueColor);
    CGContextFillRect(context, CGRectMake(0.f, 0.f, 1.f, 1.f));
    CGContextRelease(context);
    CGColorSpaceRelease(rgbColorSpace);

    red = resultingPixel[0] / 255.0f;
    green = resultingPixel[1] / 255.0f;
    blue = resultingPixel[2] / 255.0f;

    return [UIColor colorWithRed:red green:green blue:blue alpha:alpha];
}

@end


text:
using namespace foundation;
using namespace std;

namespace appleseed {
namespace studio {

MouseCoordinatesTracker::MouseCoordinatesTracker(
    RenderWidget*    widget,
    QLabel*          label)
  : m_widget(widget)
  , m_label(label)
  , m_content_width(widget->width())
  , m_content_height(widget->height())
{
    m_widget->installEventFilter(this);
}

MouseCoordinatesTracker::~MouseCoordinatesTracker()
{
    m_widget->removeEventFilter(this);
}

Vector2d MouseCoordinatesTracker::widget_to_ndc(const QPoint& point) const
{
    return
        Vector2d(
            static_cast<double>(point.x()) / m_widget->width(),
            static_cast<double>(point.y()) / m_widget->height());
}

Vector2i MouseCoordinatesTracker::widget_to_pixel(const QPoint& point) const
{
    const Vector2d ndc = widget_to_ndc(point);

    return
        Vector2i(
            static_cast<int>(ndc.x * m_content_width),
            static_cast<int>(ndc.y * m_content_height));
}

bool MouseCoordinatesTracker::eventFilter(QObject* object, QEvent* event)
{
    assert(object == m_widget);

    switch (event->type())
    {
      case QEvent::MouseMove:
        set_label_text(static_cast<QMouseEvent*>(event)->pos());
        break;

      case QEvent::Leave:
        m_label->clear();
        break;
    }

    return QObject::eventFilter(object, event);
}

void MouseCoordinatesTracker::set_label_text(const QPoint& point) const
{
    const Vector2i pix = widget_to_pixel(point);
    const Vector2d ndc = widget_to_ndc(point);

    m_label->setText(
        QString("Pixel: %1, %2 - NDC: %3, %4")
            .arg(QString::number(pix.x), 4, ' ')
            .arg(QString::number(pix.y), 4, ' ')
            .arg(QString::number(ndc.x, 'f', 5))
            .arg(QString::number(ndc.y, 'f', 5)));
}

}   // namespace studio
}   // namespace appleseed


text:
package http

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"net/http"
	"net/http/httptest"
	"path"
	"testing"

	"github.com/influxdata/httprouter"
	"github.com/influxdata/influxdb/v2"
	pcontext "github.com/influxdata/influxdb/v2/context"
	kithttp "github.com/influxdata/influxdb/v2/kit/transport/http"
	"github.com/influxdata/influxdb/v2/kv"
	"github.com/influxdata/influxdb/v2/mock"
	"github.com/influxdata/influxdb/v2/notification/endpoint"
	"github.com/influxdata/influxdb/v2/pkg/testttp"
	influxTesting "github.com/influxdata/influxdb/v2/testing"
	"go.uber.org/zap/zaptest"
)

// NewMockNotificationEndpointBackend returns a NotificationEndpointBackend with mock services.
func NewMockNotificationEndpointBackend(t *testing.T) *NotificationEndpointBackend {
	return &NotificationEndpointBackend{
		log:                         zaptest.NewLogger(t),
		HTTPErrorHandler:            kithttp.ErrorHandler(0),
		NotificationEndpointService: &mock.NotificationEndpointService{},
		UserResourceMappingService:  mock.NewUserResourceMappingService(),
		LabelService:                mock.NewLabelService(),
		UserService:                 mock.NewUserService(),
		OrganizationService:         mock.NewOrganizationService(),
	}
}

func TestService_handleGetNotificationEndpoints(t *testing.T) {
	type fields struct {
		NotificationEndpointService influxdb.NotificationEndpointService
		LabelService                influxdb.LabelService
	}
	type args struct {
		queryParams map[string][]string
	}
	type wants struct {
		statusCode  int
		contentType string
		body        string
	}

	tests := []struct {
		name   string
		fields fields
		args   args
		wants  wants
	}{
		{
			name: "get all notification endpoints",
			fields: fields{
				&mock.NotificationEndpointService{
					FindNotificationEndpointsF: func(ctx context.Context, filter influxdb.NotificationEndpointFilter, opts ...influxdb.FindOptions) ([]influxdb.NotificationEndpoint, int, error) {
						return []influxdb.NotificationEndpoint{
							&endpoint.Slack{
								Base: endpoint.Base{
									ID:     influxTesting.MustIDBase16Ptr("0b501e7e557ab1ed"),
									Name:   "hello",
									OrgID:  influxTesting.MustIDBase16Ptr("50f7ba1150f7ba11"),
									Status: influxdb.Active,
								},
								URL: "http://example.com",
							},
							&endpoint.HTTP{
								Base: endpoint.Base{
									ID:     influxTesting.MustIDBase16Ptr("c0175f0077a77005"),
									Name:   "example",
									OrgID:  influxTesting.MustIDBase16Ptr("7e55e118dbabb1ed"),
									Status: influxdb.Inactive,
								},
								URL:             "example.com",
								Username:        influxdb.SecretField{Key: "http-user-key"},
								Password:        influxdb.SecretField{Key: "http-password-key"},
								AuthMethod:      "basic",
								Method:          "POST",
								ContentTemplate: "template",
								Headers: map[string]string{
									"x-header-1": "header 1",
									"x-header-2": "header 2",
								},
							},
						}, 2, nil
					},
				},
				&mock.LabelService{
					FindResourceLabelsFn: func(ctx context.Context, f influxdb.LabelMappingFilter) ([]*influxdb.Label, error) {
						labels := []*influxdb.Label{
							{
								ID:   influxTesting.MustIDBase16("fc3dc670a4be9b9a"),
								Name: "label",
								Properties: map[string]string{
									"color": "fff000",
								},
							},
						}
						return labels, nil
					},
				},
			},
			args: args{
				map[string][]string{
					"limit": {"1"},
				},
			},
			wants: wants{
				statusCode:  http.StatusOK,
				contentType: "application/json; charset=utf-8",
				body: `
		{
		  "links": {
		    "self": "/api/v2/notificationEndpoints?descending=false&limit=1&offset=0",
		    "next": "/api/v2/notificationEndpoints?descending=false&limit=1&offset=1"
		  },
		  "notificationEndpoints": [
		   {
		     "createdAt": "0001-01-01T00:00:00Z",
		     "id": "0b501e7e557ab1ed",
		     "labels": [
		       {
		         "id": "fc3dc670a4be9b9a",
		         "name": "label",
		         "properties": {
		           "color": "fff000"
		         }
		       }
		     ],
		     "links": {
		       "labels": "/api/v2/notificationEndpoints/0b501e7e557ab1ed/labels",
		       "members": "/api/v2/notificationEndpoints/0b501e7e557ab1ed/members",
		       "owners": "/api/v2/notificationEndpoints/0b501e7e557ab1ed/owners",
		       "self": "/api/v2/notificationEndpoints/0b501e7e557ab1ed"
		     },
		     "name": "hello",
		     "orgID": "50f7ba1150f7ba11",
		     "status": "active",
			 "type": "slack",
			 "token": "",
		     "updatedAt": "0001-01-01T00:00:00Z",
		     "url": "http://example.com"
		   },
		   {
		     "createdAt": "0001-01-01T00:00:00Z",
		     "url": "example.com",
		     "id": "c0175f0077a77005",
		     "labels": [
		       {
		         "id": "fc3dc670a4be9b9a",
		         "name": "label",
		         "properties": {
		           "color": "fff000"
		         }
		       }
		     ],
		     "links": {
		       "labels": "/api/v2/notificationEndpoints/c0175f0077a77005/labels",
		       "members": "/api/v2/notificationEndpoints/c0175f0077a77005/members",
		       "owners": "/api/v2/notificationEndpoints/c0175f0077a77005/owners",
		       "self": "/api/v2/notificationEndpoints/c0175f0077a77005"
		     },
		     "name": "example",
			 "orgID": "7e55e118dbabb1ed",
			 "authMethod": "basic",
             "contentTemplate": "template",
			 "password": "secret: http-password-key",
			 "token":"",
  			 "method": "POST",
		     "status": "inactive",
			 "type": "http",
			 "headers": {
				"x-header-1": "header 1",
				"x-header-2": "header 2"
			 },
		     "updatedAt": "0001-01-01T00:00:00Z",
		     "username": "secret: http-user-key"
		   }
		   ]
		}`,
			},
		},
		{
			name: "get all notification endpoints when there are none",
			fields: fields{
				&mock.NotificationEndpointService{
					FindNotificationEndpointsF: func(ctx context.Context, filter influxdb.NotificationEndpointFilter, opts ...influxdb.FindOptions) ([]influxdb.NotificationEndpoint, int, error) {
						return []influxdb.NotificationEndpoint{}, 0, nil
					},
				},
				&mock.LabelService{},
			},
			args: args{
				map[string][]string{
					"limit": {"1"},
				},
			},
			wants: wants{
				statusCode:  http.StatusOK,
				contentType: "application/json; charset=utf-8",
				body: `
{
  "links": {
    "self": "/api/v2/notificationEndpoints?descending=false&limit=1&offset=0"
  },
  "notificationEndpoints": []
}`,
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			notificationEndpointBackend := NewMockNotificationEndpointBackend(t)
			notificationEndpointBackend.NotificationEndpointService = tt.fields.NotificationEndpointService
			notificationEndpointBackend.LabelService = tt.fields.LabelService
			h := NewNotificationEndpointHandler(zaptest.NewLogger(t), notificationEndpointBackend)

			r := httptest.NewRequest("GET", "http://any.url", nil)

			qp := r.URL.Query()
			for k, vs := range tt.args.queryParams {
				for _, v := range vs {
					qp.Add(k, v)
				}
			}
			r.URL.RawQuery = qp.Encode()
			r = r.WithContext(pcontext.SetAuthorizer(r.Context(), &influxdb.Session{UserID: user1ID}))

			w := httptest.NewRecorder()

			h.handleGetNotificationEndpoints(w, r)

			res := w.Result()
			content := res.Header.Get("Content-Type")
			body, _ := ioutil.ReadAll(res.Body)

			if res.StatusCode != tt.wants.statusCode {
				t.Errorf("%q. handleGetNotificationEndpoints() = %v, want %v", tt.name, res.StatusCode, tt.wants.statusCode)
			}
			if tt.wants.contentType != "" && content != tt.wants.contentType {
				t.Errorf("%q. handleGetNotificationEndpoints() = %v, want %v", tt.name, content, tt.wants.contentType)
			}
			if eq, diff, err := jsonEqual(string(body), tt.wants.body); err != nil || tt.wants.body != "" && !eq {
				t.Errorf("%q. handleGetNotificationEndpoints() = ***%v***", tt.name, diff)
			}
		})
	}
}

func TestService_handleGetNotificationEndpoint(t *testing.T) {
	type fields struct {
		NotificationEndpointService influxdb.NotificationEndpointService
	}
	type args struct {
		id string
	}
	type wants struct {
		statusCode  int
		contentType string
		body        string
	}

	tests := []struct {
		name   string
		fields fields
		args   args
		wants  wants
	}{
		{
			name: "get a notification endpoint by id",
			fields: fields{
				&mock.NotificationEndpointService{
					FindNotificationEndpointByIDF: func(ctx context.Context, id influxdb.ID) (influxdb.NotificationEndpoint, error) {
						if id == influxTesting.MustIDBase16("020f755c3c082000") {
							return &endpoint.HTTP{
								Base: endpoint.Base{
									ID:     influxTesting.MustIDBase16Ptr("020f755c3c082000"),
									OrgID:  influxTesting.MustIDBase16Ptr("020f755c3c082000"),
									Name:   "hello",
									Status: influxdb.Active,
								},
								URL:             "example.com",
								Username:        influxdb.SecretField{Key: "http-user-key"},
								Password:        influxdb.SecretField{Key: "http-password-key"},
								AuthMethod:      "basic",
								Method:          "POST",
								ContentTemplate: "template",
							}, nil
						}
						return nil, fmt.Errorf("not found")
					},
				},
			},
			args: args{
				id: "020f755c3c082000",
			},
			wants: wants{
				statusCode:  http.StatusOK,
				contentType: "application/json; charset=utf-8",
				body: `
		{
		  "links": {
		    "self": "/api/v2/notificationEndpoints/020f755c3c082000",
		    "labels": "/api/v2/notificationEndpoints/020f755c3c082000/labels",
		    "members": "/api/v2/notificationEndpoints/020f755c3c082000/members",
		    "owners": "/api/v2/notificationEndpoints/020f755c3c082000/owners"
		  },
		  "labels": [],
		  "authMethod": "basic",
		  "method": "POST",
		  "contentTemplate": "template",
		  "createdAt": "0001-01-01T00:00:00Z",
		  "updatedAt": "0001-01-01T00:00:00Z",
		  "id": "020f755c3c082000",
		  "url": "example.com",
		  "username": "secret: http-user-key",
		  "password": "secret: http-password-key",
		  "token":"",
		  "status": "active",
          "type": "http",
		  "orgID": "020f755c3c082000",
		  "name": "hello"
		}
		`,
			},
		},
		{
			name: "not found",
			fields: fields{
				&mock.NotificationEndpointService{
					FindNotificationEndpointByIDF: func(ctx context.Context, id influxdb.ID) (influxdb.NotificationEndpoint, error) {
						return nil, &influxdb.Error{
							Code: influxdb.ENotFound,
							Msg:  "notification endpoint not found",
						}
					},
				},
			},
			args: args{
				id: "020f755c3c082000",
			},
			wants: wants{
				statusCode: http.StatusNotFound,
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			notificationEndpointBackend := NewMockNotificationEndpointBackend(t)
			notificationEndpointBackend.HTTPErrorHandler = kithttp.ErrorHandler(0)
			notificationEndpointBackend.NotificationEndpointService = tt.fields.NotificationEndpointService
			h := NewNotificationEndpointHandler(zaptest.NewLogger(t), notificationEndpointBackend)

			r := httptest.NewRequest("GET", "http://any.url", nil)

			r = r.WithContext(context.WithValue(
				context.Background(),
				httprouter.ParamsKey,
				httprouter.Params{
					{
						Key:   "id",
						Value: tt.args.id,
					},
				}))

			w := httptest.NewRecorder()

			h.handleGetNotificationEndpoint(w, r)

			res := w.Result()
			content := res.Header.Get("Content-Type")
			body, _ := ioutil.ReadAll(res.Body)
			t.Logf(res.Header.Get("X-Influx-Error"))

			if res.StatusCode != tt.wants.statusCode {
				t.Errorf("%q. handleGetNotificationEndpoint() = %v, want %v", tt.name, res.StatusCode, tt.wants.statusCode)
			}
			if tt.wants.contentType != "" && content != tt.wants.contentType {
				t.Errorf("%q. handleGetNotificationEndpoint() = %v, want %v", tt.name, content, tt.wants.contentType)
			}
			if tt.wants.body != "" {
				if eq, diff, err := jsonEqual(string(body), tt.wants.body); err != nil {
					t.Errorf("%q, handleGetNotificationEndpoint(). error unmarshaling json %v", tt.name, err)
				} else if !eq {
					t.Errorf("%q. handleGetNotificationEndpoint() = ***%s***", tt.name, diff)
				}
			}
		})
	}
}

func TestService_handlePostNotificationEndpoint(t *testing.T) {
	type fields struct {
		Secrets                     map[string]string
		SecretService               influxdb.SecretService
		NotificationEndpointService influxdb.NotificationEndpointService
		OrganizationService         influxdb.OrganizationService
	}
	type args struct {
		endpoint interface{}
	}
	type wants struct {
		statusCode  int
		contentType string
		body        string
	}

	tests := []struct {
		name   string
		fields fields
		args   args
		wants  wants
	}{
		{
			name: "create a new notification endpoint",
			fields: fields{
				Secrets: map[string]string{},
				NotificationEndpointService: &mock.NotificationEndpointService{
					CreateNotificationEndpointF: func(ctx context.Context, edp influxdb.NotificationEndpoint, userID influxdb.ID) error {
						edp.SetID(influxTesting.MustIDBase16("020f755c3c082000"))
						edp.BackfillSecretKeys()
						return nil
					},
				},
				OrganizationService: &mock.OrganizationService{
					FindOrganizationF: func(ctx context.Context, f influxdb.OrganizationFilter) (*influxdb.Organization, error) {
						return &influxdb.Organization{ID: influxTesting.MustIDBase16("6f626f7274697320")}, nil
					},
				},
			},
			args: args{
				endpoint: map[string]interface{}{
					"name":            "hello",
					"type":            "http",
					"orgID":           "6f626f7274697320",
					"description":     "desc1",
					"status":          "active",
					"url":             "example.com",
					"username":        "user1",
					"password":        "password1",
					"authMethod":      "basic",
					"method":          "POST",
					"contentTemplate": "template",
				},
			},
			wants: wants{
				statusCode:  http.StatusCreated,
				contentType: "application/json; charset=utf-8",
				body: `
{
  "links": {
    "self": "/api/v2/notificationEndpoints/020f755c3c082000",
    "labels": "/api/v2/notificationEndpoints/020f755c3c082000/labels",
    "members": "/api/v2/notificationEndpoints/020f755c3c082000/members",
    "owners": "/api/v2/notificationEndpoints/020f755c3c082000/owners"
  },
  "url": "example.com",
  "status": "active",
  "username": "secret: 020f755c3c082000-username",
  "password": "secret: 020f755c3c082000-password",
  "token":"",
  "authMethod": "basic",
  "contentTemplate": "template",
  "type": "http",
  "method": "POST",
  "createdAt": "0001-01-01T00:00:00Z",
  "updatedAt": "0001-01-01T00:00:00Z",
  "id": "020f755c3c082000",
  "orgID": "6f626f7274697320",
  "name": "hello",
  "description": "desc1",
  "labels": []
}
`,
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			notificationEndpointBackend := NewMockNotificationEndpointBackend(t)
			notificationEndpointBackend.NotificationEndpointService = tt.fields.NotificationEndpointService
			notificationEndpointBackend.OrganizationService = tt.fields.OrganizationService

			testttp.
				PostJSON(t, prefixNotificationEndpoints, tt.args.endpoint).
				WrapCtx(authCtxFn(user1ID)).
				Do(NewNotificationEndpointHandler(zaptest.NewLogger(t), notificationEndpointBackend)).
				ExpectHeader("Content-Type", tt.wants.contentType).
				ExpectStatus(tt.wants.statusCode).
				ExpectBody(func(body *bytes.Buffer) {
					if tt.wants.body != "" {
						if eq, diff, err := jsonEqual(body.String(), tt.wants.body); err != nil {
							t.Errorf("%q, handlePostNotificationEndpoint(). error unmarshaling json %v", tt.name, err)
						} else if !eq {
							t.Errorf("%q. handlePostNotificationEndpoint() = ***%s***", tt.name, diff)
						}
					}
				})

		})
	}
}

func TestService_handleDeleteNotificationEndpoint(t *testing.T) {
	type fields struct {
		NotificationEndpointService influxdb.NotificationEndpointService
	}
	type args struct {
		id string
	}
	type wants struct {
		statusCode int
		body       string
	}

	tests := []struct {
		name   string
		fields fields
		args   args
		wants  wants
	}{
		{
			name: "remove a notification endpoint by id",
			fields: fields{
				NotificationEndpointService: &mock.NotificationEndpointService{
					DeleteNotificationEndpointF: func(ctx context.Context, id influxdb.ID) ([]influxdb.SecretField, influxdb.ID, error) {
						if id == influxTesting.MustIDBase16("020f755c3c082000") {
							return []influxdb.SecretField{
								{Key: "k1"},
							}, influxTesting.MustIDBase16("020f755c3c082001"), nil
						}

						return nil, 0, fmt.Errorf("wrong id")
					},
				},
			},
			args: args{
				id: "020f755c3c082000",
			},
			wants: wants{
				statusCode: http.StatusNoContent,
			},
		},
		{
			name: "notification endpoint not found",
			fields: fields{
				NotificationEndpointService: &mock.NotificationEndpointService{
					DeleteNotificationEndpointF: func(ctx context.Context, id influxdb.ID) ([]influxdb.SecretField, influxdb.ID, error) {
						return nil, 0, &influxdb.Error{
							Code: influxdb.ENotFound,
							Msg:  "notification endpoint not found",
						}
					},
				},
			},
			args: args{
				id: "020f755c3c082000",
			},
			wants: wants{
				statusCode: http.StatusNotFound,
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			notificationEndpointBackend := NewMockNotificationEndpointBackend(t)
			notificationEndpointBackend.HTTPErrorHandler = kithttp.ErrorHandler(0)
			notificationEndpointBackend.NotificationEndpointService = tt.fields.NotificationEndpointService

			testttp.
				Delete(t, path.Join(prefixNotificationEndpoints, tt.args.id)).
				Do(NewNotificationEndpointHandler(zaptest.NewLogger(t), notificationEndpointBackend)).
				ExpectStatus(tt.wants.statusCode).
				ExpectBody(func(buf *bytes.Buffer) {
					if tt.wants.body != "" {
						if eq, diff, err := jsonEqual(buf.String(), tt.wants.body); err != nil {
							t.Errorf("%q, handleDeleteNotificationEndpoint(). error unmarshaling json %v", tt.name, err)
						} else if !eq {
							t.Errorf("%q. handleDeleteNotificationEndpoint() = ***%s***", tt.name, diff)
						}
					}
				})
		})
	}
}

func TestService_handlePatchNotificationEndpoint(t *testing.T) {
	type fields struct {
		NotificationEndpointService influxdb.NotificationEndpointService
	}
	type args struct {
		id   string
		name string
	}
	type wants struct {
		statusCode  int
		contentType string
		body        string
	}

	tests := []struct {
		name   string
		fields fields
		args   args
		wants  wants
	}{
		{
			name: "update a notification endpoint name",
			fields: fields{
				&mock.NotificationEndpointService{
					PatchNotificationEndpointF: func(ctx context.Context, id influxdb.ID, upd influxdb.NotificationEndpointUpdate) (influxdb.NotificationEndpoint, error) {
						if id == influxTesting.MustIDBase16("020f755c3c082000") {
							d := &endpoint.Slack{
								Base: endpoint.Base{
									ID:     influxTesting.MustIDBase16Ptr("020f755c3c082000"),
									Name:   "hello",
									OrgID:  influxTesting.MustIDBase16Ptr("020f755c3c082000"),
									Status: influxdb.Active,
								},
								URL: "http://example.com",
							}

							if upd.Name != nil {
								d.Name = *upd.Name
							}

							return d, nil
						}

						return nil, fmt.Errorf("not found")
					},
				},
			},
			args: args{
				id:   "020f755c3c082000",
				name: "example",
			},
			wants: wants{
				statusCode:  http.StatusOK,
				contentType: "application/json; charset=utf-8",
				body: `
		{
		  "links": {
		    "self": "/api/v2/notificationEndpoints/020f755c3c082000",
		    "labels": "/api/v2/notificationEndpoints/020f755c3c082000/labels",
		    "members": "/api/v2/notificationEndpoints/020f755c3c082000/members",
		    "owners": "/api/v2/notificationEndpoints/020f755c3c082000/owners"
		  },
		  "createdAt": "0001-01-01T00:00:00Z",
		  "updatedAt": "0001-01-01T00:00:00Z",
		  "id": "020f755c3c082000",
		  "orgID": "020f755c3c082000",
		  "url": "http://example.com",
		  "name": "example",
		  "status": "active",
		  "type": "slack",
		  "token": "",
		  "labels": []
		}
		`,
			},
		},
		{
			name: "notification endpoint not found",
			fields: fields{
				&mock.NotificationEndpointService{
					PatchNotificationEndpointF: func(ctx context.Context, id influxdb.ID, upd influxdb.NotificationEndpointUpdate) (influxdb.NotificationEndpoint, error) {
						return nil, &influxdb.Error{
							Code: influxdb.ENotFound,
							Msg:  "notification endpoint not found",
						}
					},
				},
			},
			args: args{
				id:   "020f755c3c082000",
				name: "hello",
			},
			wants: wants{
				statusCode: http.StatusNotFound,
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			notificationEndpointBackend := NewMockNotificationEndpointBackend(t)
			notificationEndpointBackend.HTTPErrorHandler = kithttp.ErrorHandler(0)
			notificationEndpointBackend.NotificationEndpointService = tt.fields.NotificationEndpointService
			h := NewNotificationEndpointHandler(zaptest.NewLogger(t), notificationEndpointBackend)

			upd := influxdb.NotificationEndpointUpdate{}
			if tt.args.name != "" {
				upd.Name = &tt.args.name
			}

			b, err := json.Marshal(upd)
			if err != nil {
				t.Fatalf("failed to unmarshal notification endpoint update: %v", err)
			}

			r := httptest.NewRequest("GET", "http://any.url", bytes.NewReader(b))
			r = r.WithContext(pcontext.SetAuthorizer(r.Context(), &influxdb.Session{UserID: user1ID}))

			r = r.WithContext(context.WithValue(
				context.Background(),
				httprouter.ParamsKey,
				httprouter.Params{
					{
						Key:   "id",
						Value: tt.args.id,
					},
				}))

			w := httptest.NewRecorder()

			h.handlePatchNotificationEndpoint(w, r)

			res := w.Result()
			content := res.Header.Get("Content-Type")
			body, _ := ioutil.ReadAll(res.Body)

			if res.StatusCode != tt.wants.statusCode {
				t.Errorf("%q. handlePatchNotificationEndpoint() = %v, want %v %v", tt.name, res.StatusCode, tt.wants.statusCode, w.Header())
			}
			if tt.wants.contentType != "" && content != tt.wants.contentType {
				t.Errorf("%q. handlePatchNotificationEndpoint() = %v, want %v", tt.name, content, tt.wants.contentType)
			}
			if tt.wants.body != "" {
				if eq, diff, err := jsonEqual(string(body), tt.wants.body); err != nil {
					t.Errorf("%q, handlePatchNotificationEndpoint(). error unmarshaling json %v", tt.name, err)
				} else if !eq {
					t.Errorf("%q. handlePatchNotificationEndpoint() = ***%s***", tt.name, diff)
				}
			}
		})
	}
}

func TestService_handleUpdateNotificationEndpoint(t *testing.T) {
	type fields struct {
		NotificationEndpointService influxdb.NotificationEndpointService
	}
	type args struct {
		id  string
		edp map[string]interface{}
	}
	type wants struct {
		statusCode  int
		contentType string
		body        string
	}

	tests := []struct {
		name   string
		fields fields
		args   args
		wants  wants
	}{
		{
			name: "update a notification endpoint name",
			fields: fields{
				NotificationEndpointService: &mock.NotificationEndpointService{
					UpdateNotificationEndpointF: func(ctx context.Context, id influxdb.ID, edp influxdb.NotificationEndpoint, userID influxdb.ID) (influxdb.NotificationEndpoint, error) {
						if id == influxTesting.MustIDBase16("020f755c3c082000") {
							edp.SetID(id)
							edp.BackfillSecretKeys()
							return edp, nil
						}

						return nil, fmt.Errorf("not found")
					},
				},
			},
			args: args{
				id: "020f755c3c082000",
				edp: map[string]interface{}{
					"name":   "example",
					"status": "active",
					"orgID":  "020f755c3c082001",
					"type":   "slack",
					"token":  "",
					"url":    "example.com",
				},
			},
			wants: wants{
				statusCode:  http.StatusOK,
				contentType: "application/json; charset=utf-8",
				body: `
		{
		  "links": {
		    "self": "/api/v2/notificationEndpoints/020f755c3c082000",
		    "labels": "/api/v2/notificationEndpoints/020f755c3c082000/labels",
		    "members": "/api/v2/notificationEndpoints/020f755c3c082000/members",
		    "owners": "/api/v2/notificationEndpoints/020f755c3c082000/owners"
		  },
		  "createdAt": "0001-01-01T00:00:00Z",
		  "updatedAt": "0001-01-01T00:00:00Z",
		  "id": "020f755c3c082000",
		  "orgID": "020f755c3c082001",
		  "name": "example",
		  "url": "example.com",
          "type": "slack",
		  "status": "active",
		  "token": "",
          "labels": []
		}
		`,
			},
		},
		{
			name: "notification endpoint not found",
			fields: fields{
				NotificationEndpointService: &mock.NotificationEndpointService{
					UpdateNotificationEndpointF: func(ctx context.Context, id influxdb.ID, edp influxdb.NotificationEndpoint, userID influxdb.ID) (influxdb.NotificationEndpoint, error) {
						return nil, &influxdb.Error{
							Code: influxdb.ENotFound,
							Msg:  "notification endpoint not found",
						}
					},
				},
			},
			args: args{
				id: "020f755c3c082000",
				edp: map[string]interface{}{
					"type": "slack",
					"name": "example",
				},
			},
			wants: wants{
				statusCode: http.StatusNotFound,
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			notificationEndpointBackend := NewMockNotificationEndpointBackend(t)
			notificationEndpointBackend.HTTPErrorHandler = kithttp.ErrorHandler(0)
			notificationEndpointBackend.NotificationEndpointService = tt.fields.NotificationEndpointService

			resp := testttp.
				PutJSON(t, path.Join(prefixNotificationEndpoints, tt.args.id), tt.args.edp).
				WrapCtx(authCtxFn(user1ID)).
				Do(NewNotificationEndpointHandler(zaptest.NewLogger(t), notificationEndpointBackend)).
				ExpectStatus(tt.wants.statusCode).
				ExpectBody(func(body *bytes.Buffer) {
					if tt.wants.body != "" {
						if eq, diff, err := jsonEqual(body.String(), tt.wants.body); err != nil {
							t.Errorf("%q, handlePutNotificationEndpoint(). error unmarshaling json %v", tt.name, err)
						} else if !eq {
							t.Errorf("%q. handlePutNotificationEndpoint() = ***%s***", tt.name, diff)
						}
					}
				})
			if tt.wants.contentType != "" {
				resp.ExpectHeader("Content-Type", tt.wants.contentType)
			}
		})
	}
}

func TestService_handlePostNotificationEndpointMember(t *testing.T) {
	type fields struct {
		UserService influxdb.UserService
	}
	type args struct {
		notificationEndpointID string
		user                   *influxdb.User
	}
	type wants struct {
		statusCode  int
		contentType string
		body        string
	}

	tests := []struct {
		name   string
		fields fields
		args   args
		wants  wants
	}{
		{
			name: "add a notification endpoint member",
			fields: fields{
				UserService: &mock.UserService{
					FindUserByIDFn: func(ctx context.Context, id influxdb.ID) (*influxdb.User, error) {
						return &influxdb.User{
							ID:     id,
							Name:   "name",
							Status: influxdb.Active,
						}, nil
					},
				},
			},
			args: args{
				notificationEndpointID: "020f755c3c082000",
				user: &influxdb.User{
					ID: influxTesting.MustIDBase16("6f626f7274697320"),
				},
			},
			wants: wants{
				statusCode:  http.StatusCreated,
				contentType: "application/json; charset=utf-8",
				body: `
{
  "links": {
    "logs": "/api/v2/users/6f626f7274697320/logs",
    "self": "/api/v2/users/6f626f7274697320"
  },
  "role": "member",
  "id": "6f626f7274697320",
	"name": "name",
	"status": "active"
}
`,
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			notificationEndpointBackend := NewMockNotificationEndpointBackend(t)
			notificationEndpointBackend.UserService = tt.fields.UserService
			h := NewNotificationEndpointHandler(zaptest.NewLogger(t), notificationEndpointBackend)

			b, err := json.Marshal(tt.args.user)
			if err != nil {
				t.Fatalf("failed to marshal user: %v", err)
			}

			path := fmt.Sprintf("/api/v2/notificationEndpoints/%s/members", tt.args.notificationEndpointID)
			r := httptest.NewRequest("POST", path, bytes.NewReader(b))
			r = r.WithContext(pcontext.SetAuthorizer(r.Context(), &influxdb.Session{UserID: user1ID}))
			w := httptest.NewRecorder()

			h.ServeHTTP(w, r)

			res := w.Result()
			content := res.Header.Get("Content-Type")
			body, _ := ioutil.ReadAll(res.Body)

			if res.StatusCode != tt.wants.statusCode {
				t.Errorf("%q. handlePostNotificationEndpointMember() = %v, want %v", tt.name, res.StatusCode, tt.wants.statusCode)
			}
			if tt.wants.contentType != "" && content != tt.wants.contentType {
				t.Errorf("%q. handlePostNotificationEndpointMember() = %v, want %v", tt.name, content, tt.wants.contentType)
			}
			if eq, diff, err := jsonEqual(string(body), tt.wants.body); err != nil {
				t.Errorf("%q, handlePostNotificationEndpointMember(). error unmarshaling json %v", tt.name, err)
			} else if tt.wants.body != "" && !eq {
				t.Errorf("%q. handlePostNotificationEndpointMember() = ***%s***", tt.name, diff)
			}
		})
	}
}

func TestService_handlePostNotificationEndpointOwner(t *testing.T) {
	type fields struct {
		UserService influxdb.UserService
	}
	type args struct {
		notificationEndpointID string
		user                   *influxdb.User
	}
	type wants struct {
		statusCode  int
		contentType string
		body        string
	}

	cases := []struct {
		name   string
		fields fields
		args   args
		wants  wants
	}{
		{
			name: "add a notification endpoint owner",
			fields: fields{
				UserService: &mock.UserService{
					FindUserByIDFn: func(ctx context.Context, id influxdb.ID) (*influxdb.User, error) {
						return &influxdb.User{
							ID:     id,
							Name:   "name",
							Status: influxdb.Active,
						}, nil
					},
				},
			},
			args: args{
				notificationEndpointID: "020f755c3c082000",
				user: &influxdb.User{
					ID: influxTesting.MustIDBase16("6f626f7274697320"),
				},
			},
			wants: wants{
				statusCode:  http.StatusCreated,
				contentType: "application/json; charset=utf-8",
				body: `
{
  "links": {
    "logs": "/api/v2/users/6f626f7274697320/logs",
    "self": "/api/v2/users/6f626f7274697320"
  },
  "role": "owner",
  "id": "6f626f7274697320",
	"name": "name",
	"status": "active"
}
`,
			},
		},
	}

	for _, tt := range cases {
		t.Run(tt.name, func(t *testing.T) {
			notificationEndpointBackend := NewMockNotificationEndpointBackend(t)
			notificationEndpointBackend.UserService = tt.fields.UserService
			h := NewNotificationEndpointHandler(zaptest.NewLogger(t), notificationEndpointBackend)

			b, err := json.Marshal(tt.args.user)
			if err != nil {
				t.Fatalf("failed to marshal user: %v", err)
			}

			path := fmt.Sprintf("/api/v2/notificationEndpoints/%s/owners", tt.args.notificationEndpointID)
			r := httptest.NewRequest("POST", path, bytes.NewReader(b))
			w := httptest.NewRecorder()

			h.ServeHTTP(w, r)

			res := w.Result()
			content := res.Header.Get("Content-Type")
			body, _ := ioutil.ReadAll(res.Body)

			if res.StatusCode != tt.wants.statusCode {
				t.Errorf("%q. handlePostNotificationEndpointOwner() = %v, want %v", tt.name, res.StatusCode, tt.wants.statusCode)
			}
			if tt.wants.contentType != "" && content != tt.wants.contentType {
				t.Errorf("%q. handlePostNotificationEndpointOwner() = %v, want %v", tt.name, content, tt.wants.contentType)
			}
			if eq, diff, err := jsonEqual(string(body), tt.wants.body); err != nil {
				t.Errorf("%q, handlePostNotificationEndpointOwner(). error unmarshaling json %v", tt.name, err)
			} else if tt.wants.body != "" && !eq {
				t.Errorf("%q. handlePostNotificationEndpointOwner() = ***%s***", tt.name, diff)
			}
		})
	}
}

func initNotificationEndpointService(f influxTesting.NotificationEndpointFields, t *testing.T) (influxdb.NotificationEndpointService, influxdb.SecretService, func()) {
	ctx := context.Background()
	store := NewTestInmemStore(t)
	logger := zaptest.NewLogger(t)
	svc := kv.NewService(logger, store)
	svc.IDGenerator = f.IDGenerator
	svc.TimeGenerator = f.TimeGenerator

	for _, v := range f.Orgs {
		if err := svc.PutOrganization(ctx, v); err != nil {
			t.Fatalf("failed to replace org: %v", err)
		}
	}

	for _, m := range f.UserResourceMappings {
		if err := svc.CreateUserResourceMapping(ctx, m); err != nil {
			t.Fatalf("failed to populate user resource mapping: %v", err)
		}
	}

	for _, v := range f.NotificationEndpoints {
		if err := svc.PutNotificationEndpoint(ctx, v); err != nil {
			t.Fatalf("failed to update endpoint: %v", err)
		}
	}

	fakeBackend := NewMockNotificationEndpointBackend(t)
	fakeBackend.NotificationEndpointService = svc
	fakeBackend.UserService = svc
	fakeBackend.UserResourceMappingService = svc
	fakeBackend.OrganizationService = svc

	handler := NewNotificationEndpointHandler(zaptest.NewLogger(t), fakeBackend)
	auth := func(next http.Handler) http.HandlerFunc {
		return func(w http.ResponseWriter, r *http.Request) {
			r = r.WithContext(pcontext.SetAuthorizer(r.Context(), &influxdb.Session{UserID: user1ID}))
			next.ServeHTTP(w, r)
		}
	}
	server := httptest.NewServer(auth(handler))
	done := server.Close

	client := mustNewHTTPClient(t, server.URL, "")
	return NewNotificationEndpointService(client), svc, done
}

func TestNotificationEndpointService(t *testing.T) {
	t.Skip("wonky")

	tests := []struct {
		name   string
		testFn func(init func(influxTesting.NotificationEndpointFields, *testing.T) (influxdb.NotificationEndpointService, influxdb.SecretService, func()), t *testing.T)
	}{
		{
			name:   "CreateNotificationEndpoint",
			testFn: influxTesting.CreateNotificationEndpoint,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tt.testFn(initNotificationEndpointService, t)
		})
	}
}

func authCtxFn(userID influxdb.ID) func(context.Context) context.Context {
	return func(ctx context.Context) context.Context {
		return pcontext.SetAuthorizer(ctx, &influxdb.Session{UserID: userID})
	}
}


text:
__author__    = "Andre Merzky, Ole Weidner"
__copyright__ = "Copyright 2012-2013, The SAGA Project"
__license__   = "MIT"

""" Local filesystem adaptor implementation """

import os
import pprint
import shutil
import traceback

import saga.url
import saga.adaptors.base
import saga.adaptors.cpi.filesystem

import saga.utils.misc

SYNC_CALL  = saga.adaptors.cpi.decorators.SYNC_CALL
ASYNC_CALL = saga.adaptors.cpi.decorators.ASYNC_CALL

###############################################################################
# adaptor info
#

_ADAPTOR_NAME          = 'saga.adaptor.filesystem.local'
_ADAPTOR_SCHEMAS       = ['file', 'local']
_ADAPTOR_OPTIONS       = []
_ADAPTOR_CAPABILITIES  = {}

_ADAPTOR_DOC           = {
    'name'             : _ADAPTOR_NAME,
    'cfg_options'      : _ADAPTOR_OPTIONS,
    'capabilities'     : _ADAPTOR_CAPABILITIES,
    'description'      : """This adaptor interacts with local filesystem, by
                            using the (POSIX like) os and shutil Python packages.""",
    'schemas'          : {'file'  : 'local filesystem.',
                          'local' : 'alias for *file*'
    },
}

_ADAPTOR_INFO          = {
    'name'             : _ADAPTOR_NAME,
    'version'          : 'v0.2',
    'schemas'          : _ADAPTOR_SCHEMAS,
    'cpis'             : [{
        'type'         : 'saga.filesystem.Directory',
        'class'        : 'LocalDirectory'
        },
        {
        'type'         : 'saga.filesystem.File',
        'class'        : 'LocalFile'
        }
    ]
}

###############################################################################
# The adaptor class

class Adaptor (saga.adaptors.base.Base):
    """
    This is the actual adaptor class, which gets loaded by SAGA (i.e. by the
    SAGA engine), and which registers the CPI implementation classes which
    provide the adaptor's functionality.
    """

    def __init__ (self) :

        saga.adaptors.base.Base.__init__ (self, _ADAPTOR_INFO, _ADAPTOR_OPTIONS)

        # the adaptor *singleton* creates a (single) instance of a bulk handler
        # (BulkDirectory), which implements container_* bulk methods.
        self._bulk = BulkDirectory ()

    def sanity_check (self) :
        # nothing to check for, local file system should always be accessible
        pass

###############################################################################
#
class BulkDirectory (saga.adaptors.cpi.filesystem.Directory) :
    """
    Well, this implementation can handle bulks, but cannot optimize them.
    We leave that code here anyway, for demonstration -- but those methods
    are also provided as fallback, and are thusly used if the adaptor does
    not implement the bulk container_* methods at all.
    """

    def __init__ (self) :
        pass

    def container_wait (self, tasks, mode, timeout) :
      # print " ~ bulk wait ~~~~~~~~~~~~~~~~~~~~~ "
      # pprint.pprint (tasks)
        if timeout >= 0 :
            raise saga.exceptions.BadParameter ("Cannot handle timeouts > 0")
        for task in tasks :
            task.wait ()
      # print " ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ "

    def container_cancel (self, tasks) :
      # print " ~ bulk wait ~~~~~~~~~~~~~~~~~~~~~ "
      # pprint.pprint (tasks)
        for task in tasks :
            task.cancel ()
      # print " ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ "

    def container_copy (self, tasks) :
        """
        A *good* implementation would dig the file copy operations from the
        tasks, and run them in a bulk -- we can't do that, so simply *run* the
        individual tasks, falling back to the default non-bulk asynchronous copy
        operation...
        """
      # print " ~ bulk copy ~~~~~~~~~~~~~~~~~~~~~ "
      # pprint.pprint (tasks)
        for task in tasks :
            task.run ()
      # print " ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ "

    # the container methods for the other calls are obviously similar, and left
    # out here.  The :class:`saga.task.Container` implementation will fall back
    # to the non-bulk async calls for all then.

###############################################################################
#
class LocalDirectory (saga.adaptors.cpi.filesystem.Directory) :

    def __init__ (self, api, adaptor) :

        _cpi_base = super  (LocalDirectory, self)
        _cpi_base.__init__ (api, adaptor)

    @SYNC_CALL
    def init_instance (self, adaptor_state, url, flags, session) :

        self._url       = url
        self._flags     = flags
        self._session   = session
        self._container = self._adaptor._bulk

        self._init_check ()

        return self.get_api ()

    @ASYNC_CALL
    def init_instance_async (self, adaptor_state, url, flags, session, ttype) :

        self._url     = url
        self._flags   = flags
        self._session = session

        c = { 'url'     : self._url,
              'flags'   : self._flags,
              'session' : self._session }

        return saga.task.Task (self, 'init_instance', c, ttype)

    def _init_check (self) :

        url   = self._url
        flags = self._flags

        if not saga.utils.misc.url_is_local (url) :
            raise saga.exceptions.BadParameter ("Cannot handle url %s (not local)"     %  url)
        if url.port :
            raise saga.exceptions.BadParameter ("Cannot handle url %s (has port)"      %  url)
        if url.fragment :
            raise saga.exceptions.BadParameter ("Cannot handle url %s (has fragment)"  %  url)
        if url.query :
            raise saga.exceptions.BadParameter ("Cannot handle url %s (has query)"     %  url)
        if url.username :
            raise saga.exceptions.BadParameter ("Cannot handle url %s (has username)"  %  url)
        if url.password :
            raise saga.exceptions.BadParameter ("Cannot handle url %s (has password)"  %  url)

        self._path = url.path
        path       = url.path

        if not os.path.exists (path) :

            if saga.filesystem.CREATE & flags :
                if saga.filesystem.CREATE_PARENTS & flags :
                    try :
                        os.makedirs (path)
                    except Exception as e :
                        raise saga.exceptions.NoSuccess ("Could not 'mkdir -p %s': %s)"  \
                                                        % (path, str(e)))
                else :
                    try :
                        os.mkdir (path)
                    except Exception as e :
                        raise saga.exceptions.NoSuccess ("Could not 'mkdir %s': %s)"  \
                                                        % (path, str(e)))
            else :
                raise saga.exceptions.BadParameter ("Cannot handle url %s (directory does not exist)"  \
                                                   %  path)

        if not os.path.isdir (path) :
            raise saga.exceptions.BadParameter ("Cannot handle url %s (is not a directory)"  \
                                               %  path)

    @SYNC_CALL
    def get_url (self) :

        return self._url

    @SYNC_CALL
    def open (self, url, flags) :

      # print "sync open: - '%s' - '%s' - "  %  (url, flags)

        if not url.scheme and not url.host :
            url = saga.url.Url (str(self._url) + '/' + str(url))

        return saga.filesystem.File (url, flags, self._session, _adaptor=self._adaptor)

    # # the async call implementation works, in pair with task_run.  As it is
    # disabled though, the threaded async fallback from the cpi layer will be
    # used.
    # @ASYNC_CALL
    # def open_async (self, url, flags, ttype) :
    #
    #     c = { 'url'     : url,
    #             'flags'   : flags}
    #
    #     return saga.task.Task (self, 'open', c, ttype)

    @SYNC_CALL
    def copy (self, source, target, flags) :

        src_url = saga.url.Url (source)
        src     = src_url.path
        tgt_url = saga.url.Url (target)
        tgt     = tgt_url.path

        if src_url.schema :
            if not src_url.schema.lower () in _ADAPTOR_SCHEMAS :
                raise saga.exceptions.BadParameter ("Cannot handle url %s (not local)" %  source)

        if not saga.utils.misc.url_is_local (src_url) :
            raise saga.exceptions.BadParameter ("Cannot handle url %s (not local)"     %  source)

        if src[0] != '/' :
            if not src_url.scheme and not src_url.host :
                if self._url.path[0] == '/' :
                    src = "%s/%s"   % (self._url.path, src)

        if tgt_url.schema :
            if not tgt_url.schema.lower () in _ADAPTOR_SCHEMAS :
                raise saga.exceptions.BadParameter ("Cannot handle url %s (not local)" %  target)

        if not saga.utils.misc.url_is_local (tgt_url) :
            raise saga.exceptions.BadParameter ("Cannot handle url %s (not local)"     %  target)

        if tgt[0] != '/' :
            if not tgt_url.scheme and not tgt_url.host :
                if self._url.path[0] == '/' :
                    tgt = "%s/%s"   % (self._url.path, tgt)

        # FIXME: eval flags, check for existence, etc.

        if os.path.isdir (src) :
          # print "sync copy tree %s -> %s" % (src, tgt)
            shutil.copytree (src, tgt)

        else :
          # print "sync copy %s -> %s" % (src, tgt)
            shutil.copy2 (src, tgt)

    @SYNC_CALL
    def copy_self (self, target, flags) :

        return self.copy (self._url, target, flags)

    @ASYNC_CALL
    def copy_async (self, src, tgt, flags, ttype) :

      # print "async copy %s -> %s [%s]" % (src, tgt, ttype)

        c = { 'src'     : src,
              'tgt'     : tgt,
              'flags'   : flags }

        return saga.task.Task (self, 'copy', c, ttype)

    def task_wait (self, task, timout) :
        # FIXME: our task_run moves all tasks into DONE state... :-/
        pass

    def task_run (self, task) :
        # FIXME: that should be generalized, possibly wrapped into a thread, and
        # moved to CPI level

        call = task._method_type
        c    = task._method_context

        if call == 'copy' :
            try :
                task._set_result (self.copy (c['src'], c['tgt'], c['flags']))
                task._set_state  (saga.task.DONE)
            except Exception as e :
                task._set_exception (e)
                task._set_state     (saga.task.FAILED)
        elif call == 'init_instance' :
            try :
                self.init_instance ({}, c['url'], c['flags'], c['session'])
                task._set_result (self.get_api ())
                task._set_state  (saga.task.DONE)
            except Exception as e :
                task._set_exception (e)
                task._set_state     (saga.task.FAILED)
        elif call == 'open' :
            try :
                task._set_result (self.open (c['url'], c['flags']))
                task._set_state  (saga.task.DONE)
            except Exception as e :
                task._set_exception (e)
                task._set_state     (saga.task.FAILED)
        else :
            raise saga.exceptions.NotImplemented ("Cannot handle %s tasks" %  call)

######################################################################
#
# file adaptor class
#
class LocalFile (saga.adaptors.cpi.filesystem.File) :

    def __init__ (self, api, adaptor) :

        _cpi_base = super  (LocalFile, self)
        _cpi_base.__init__ (api, adaptor)

    def _dump (self) :
        print "url    : %s"  % self._url
        print "flags  : %s"  % self._flags
        print "session: %s"  % self._session

    @SYNC_CALL
    def init_instance (self, adaptor_state, url, flags, session) :

        self._url     = url
        self._flags   = flags
        self._session = session

        self._init_check ()

        return self.get_api ()

    @ASYNC_CALL
    def init_instance_async (self, adaptor_state, url, flags, session, ttype) :

        self._url     = url
        self._flags   = flags
        self._session = session

        c = { 'url'     : self._url,
              'flags'   : self._flags,
              'session' : self._session }

        t = saga.task.Task (self, 'init_instance', c, ttype)

        # FIXME: move to task_run...
        self._init_check ()

        t._set_result (saga.filesystem.File (url, flags, session, _adaptor=self._adaptor))
        t._set_state  (saga.task.DONE)

        return t

    def _init_check (self) :

        url   = self._url
        flags = self._flags

        if not saga.utils.misc.url_is_local (url) :
            raise saga.exceptions.BadParameter ("Cannot handle url %s (not local)"     %  url)
        if url.port :
            raise saga.exceptions.BadParameter ("Cannot handle url %s (has port)"      %  url)
        if url.fragment :
            raise saga.exceptions.BadParameter ("Cannot handle url %s (has fragment)"  %  url)
        if url.query :
            raise saga.exceptions.BadParameter ("Cannot handle url %s (has query)"     %  url)
        if url.username :
            raise saga.exceptions.BadParameter ("Cannot handle url %s (has username)"  %  url)
        if url.password :
            raise saga.exceptions.BadParameter ("Cannot handle url %s (has password)"  %  url)

        self._path = url.path
        path       = url.path

        if not os.path.exists (path) :

            (dirname, filename) = os.path.split (path)

            if not filename :
                raise saga.exceptions.BadParameter ("Cannot handle url %s (names directory)"  \
                                                 %  path)

            if not os.path.exists (dirname) :
                if saga.filesystem.CREATE_PARENTS & flags :
                    try :
                        os.makedirs (path)
                    except Exception as e :
                        raise saga.exceptions.NoSuccess ("Could not 'mkdir -p %s': %s)"  \
                                                        % (path, str(e)))
                else :
                    raise saga.exceptions.BadParameter ("Cannot handle url %s (parent dir does not exist)"  \
                                                     %  path)

            if not os.path.exists (filename) :
                if saga.filesystem.CREATE & flags :
                    try :
                        open (path, 'w').close () # touch
                    except Exception as e :
                        raise saga.exceptions.NoSuccess ("Could not 'touch %s': %s)"  \
                                                        % (path, str(e)))
                else :
                    raise saga.exceptions.BadParameter ("Cannot handle url %s (file does not exist)"  \
                                                     %  path)

        if not os.path.isfile (path) :
            raise saga.exceptions.BadParameter ("Cannot handle url %s (is not a file)"  \
                                               %  path)

    @SYNC_CALL
    def get_url (self) :
        return self._url

    @ASYNC_CALL
    def get_url_async (self, ttype) :

        c = {}
        t = saga.task.Task (self, 'get_url', c, ttype)

        # FIXME: move to task_run...
        t._set_state  (saga.task.DONE)
        t._set_result (self._url)

        return t

    @SYNC_CALL
    def get_size_self (self) :
        return os.path.getsize (self._url.path)

    @ASYNC_CALL
    def get_size_self_async (self, ttype) :

        c = {}
        t = saga.task.Task (self, 'get_size', c, ttype)

        # FIXME: move to task_run...
        t._set_result (os.path.getsize (self._url.path))
        t._set_state  (saga.task.DONE)

        return t

    @SYNC_CALL
    def copy_self (self, target, flags) :

        tgt_url = saga.url.Url (target)
        tgt     = tgt_url.path
        src     = self._url.path

        if tgt_url.schema :
            if not tgt_url.schema.lower () in _ADAPTOR_SCHEMAS :
                raise saga.exceptions.BadParameter ("Cannot handle url %s (not local)" %  target)

        if not saga.utils.misc.url_is_local (tgt_url) :
            raise saga.exceptions.BadParameter ("Cannot handle url %s (not local)"     %  target)

        if tgt[0] != '/' :
            tgt = "%s/%s"   % (os.path.dirname (src), tgt)

      # print " copy %s %s" % (self._url, tgt)
        shutil.copy2 (src, tgt)

    def task_run (self, task) :
        # FIXME: that should be generalized, possibly wrapped into a thread, and
        # moved to CPI level

        call = task._method_type
        c    = task._method_context

        if call == 'copy_self' :
            try :
                task._set_result (self.copy_self (c['tgt'], c['flags']))
                task._set_state  (saga.task.DONE)
            except Exception as e :
                task._set_exception (e)
                task._set_state     (saga.task.FAILED)
        elif call == 'get_size' :
            try :
                task._set_result (self.get_size_self ())
                task._set_state  (saga.task.DONE)
            except Exception as e :
                task._set_exception (e)
                task._set_state     (saga.task.FAILED)
        else :
            raise saga.exceptions.NotImplemented ("Cannot handle %s tasks" %  call)


text:
#ifndef READ_PARSERS_HH
#define READ_PARSERS_HH

#include <regex.h>
#include <stddef.h>
#include <stdint.h>
#include <cstdlib>
#include <iostream>
#include <string>
#include <utility>

#include "khmer.hh"
#include "khmer_exception.hh"

namespace khmer
{

namespace read_parsers
{

struct NoMoreReadsAvailable : public  khmer_file_exception {
    explicit NoMoreReadsAvailable(const std::string& msg) :
        khmer_file_exception(msg) {}
    NoMoreReadsAvailable() :
        khmer_file_exception("No more reads available in this stream.") {}
};

struct InvalidRead : public  khmer_value_exception {
    explicit InvalidRead(const std::string& msg) :
        khmer_value_exception(msg) {}
    InvalidRead() :
        khmer_value_exception("Invalid FASTA/Q read") {}
};

struct UnknownPairReadingMode : public  khmer_value_exception {
    explicit UnknownPairReadingMode(const std::string& msg) :
        khmer_value_exception(msg) {}
    UnknownPairReadingMode() :
        khmer_value_exception("Unknown pair reading mode supplied.") {}
};

struct InvalidReadPair : public  khmer_value_exception {
    explicit InvalidReadPair(const std::string& msg) :
        khmer_value_exception(msg) {}
    InvalidReadPair() :
        khmer_value_exception("Invalid read pair detected.") {}
};

struct Read {
    std:: string    name;
    std:: string    annotations;
    std:: string    sequence;
    std:: string    quality;
    // TODO? Add description field.

    inline void reset ( )
    {
        name.clear( );
        annotations.clear( );
        sequence.clear( );
        quality.clear( );
    }

    void write_to(std::ostream&);
};

typedef std:: pair< Read, Read >	ReadPair;

struct IParser {

    enum {
        PAIR_MODE_ALLOW_UNPAIRED = 0,
        PAIR_MODE_IGNORE_UNPAIRED,
        PAIR_MODE_ERROR_ON_UNPAIRED
    };

    static IParser * const  get_parser(
        std:: string const 	&ifile_name
    );

    IParser( );
    virtual ~IParser( );

    virtual bool		is_complete( ) = 0;

    // Note: 'get_next_read' exists for legacy reasons.
    //	     In the long term, it should be eliminated in favor of direct use of
    //	     'imprint_next_read'. A potentially costly copy-by-value happens
    //	     upon return.
    // TODO: Eliminate all calls to 'get_next_read'.
    // Or switch to C++11 w/ move constructors
    inline Read		get_next_read( )
    {
        Read the_read;
        imprint_next_read( the_read );
        return the_read;
    }
    virtual void	imprint_next_read( Read &the_read ) = 0;

    virtual void	imprint_next_read_pair(
        ReadPair &the_read_pair,
        uint8_t mode = PAIR_MODE_ERROR_ON_UNPAIRED
    );

    size_t		    get_num_reads()
    {
        return _num_reads;
    }

protected:

    size_t		_num_reads;
    bool        _have_qualities;
    regex_t		_re_read_2_nosub;
    regex_t		_re_read_1;
    regex_t		_re_read_2;

#if (0)
    void		_imprint_next_read_pair_in_allow_mode(
        ReadPair &the_read_pair
    );
#endif

    void		_imprint_next_read_pair_in_ignore_mode(
        ReadPair &the_read_pair
    );
    void		_imprint_next_read_pair_in_error_mode(
        ReadPair &the_read_pair
    );
    bool		_is_valid_read_pair(
        ReadPair &the_read_pair, regmatch_t &match_1, regmatch_t &match_2
    );

}; // struct IParser

class SeqAnParser : public IParser
{

public:
    explicit SeqAnParser( const char * filename );
    ~SeqAnParser( );

    bool is_complete( );
    void imprint_next_read(Read &the_read);

private:
    struct Handle;

    Handle* _private;

};

inline PartitionID _parse_partition_id(std::string name)
{
    PartitionID p = 0;
    const char * s = name.c_str() + name.length() - 1;
    if (!(*(s + 1) == (unsigned int) NULL)) {
        throw khmer_exception();
    }

    while(*s != '\t' && s >= name.c_str()) {
        s--;
    }

    if (*s == '\t') {
        p = (PartitionID) atoi(s + 1);
    } else {
        std::cerr << "consume_partitioned_fasta barfed on read "  << name << "\n";
        throw khmer_exception();
    }

    return p;
}

} // namespace read_parsers

} // namespace khmer

#endif // READ_PARSERS_HH

// vim: set ft=cpp sts=4 sw=4 tw=80:


text:
package org.omarket.trading.util;

import rx.Observable;
import rx.Observable.Operator;
import rx.Observer;
import rx.Producer;
import rx.Subscriber;
import rx.exceptions.Exceptions;
import rx.exceptions.MissingBackpressureException;
import rx.internal.operators.BackpressureUtils;
import rx.internal.util.RxRingBuffer;
import rx.subscriptions.CompositeSubscription;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.PriorityQueue;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Merges source {@link Observable}s into one {@code Observable}, without any
 * transformation, in the order specified by the {@code comparator} (assuming
 * the source {@link Observable}s have their items pre-sorted in the order
 * consistent with the {@code comparator}).
 *

 * Items emitted by each of the source {@link Observable}s are expected to be
 * pre-ordered according to the specified {@code comparator}, otherwise the
 * order of the resulting {@code Observable}'s items is undefined.
 *

 * The resulting observable will start emitting items only after obtaining at
 * least one item from each of the source {@link Observable}s being
 * merged.
 *

 * The operator has as a space cost of {@code O(M)}, where M is a number of the
 * source {@link Observable}s to merge, and it may hold up to
 * {@code M * RxRingBuffer.SIZE} items at a time in memory.
 *

 * This operator supports backpressure and is loosely based on
 * {@link rx.internal.operators.OperatorZip}.
 *
 * @param <T> the type of the items emitted by both the source and merged
 *            {@code Observable}s
 */
public final class OperatorMergeSorted<T> implements Operator<T, Observable<? extends T>> {

    private final Comparator<T> comparator;

    /**
     * Constructs an operator that emits its elements according to their
     * {@linkplain Comparable} natural ordering.
     */
    public OperatorMergeSorted() {
        this.comparator = null;
    }

    /**
     * Constructs an operator using the specified {@link Comparator}.
     *
     * @param comparator a comparator to use for a sorted merge.
     */
    public OperatorMergeSorted(Comparator<T> comparator) {
        this.comparator = comparator;
    }

    @Override
    public Subscriber<? super Observable<? extends T>> call(final Subscriber<? super T> child) {
        final SortedMerge<T> merger = new SortedMerge<T>(comparator, child);
        final MergeSortedProducer<T> producer = new MergeSortedProducer<T>(merger);
        child.setProducer(producer);
        return new MergeSortedSubscriber(child, merger, producer);
    }

    private static final class MergeSortedProducer<T> extends AtomicLong implements Producer {
        /** */
        private static final long serialVersionUID = -1216676403723546796L;
        private SortedMerge<T> merge;

        public MergeSortedProducer(SortedMerge<T> merge) {
            this.merge = merge;
        }

        @Override
        public void request(long n) {
            BackpressureUtils.getAndAddRequest(this, n);
            // try and claim emission if no other threads are doing so
            merge.tick();
        }

    }

    private static final class SortedMerge<T> extends AtomicLong {
        static final int THRESHOLD = (int) (RxRingBuffer.SIZE * 0.7);
        /** */
        private static final long serialVersionUID = 1237247396158074733L;
        private final Comparator<T> comparator;
        private final Observer<? super T> child;
        private final CompositeSubscription childSubscription = new CompositeSubscription();
        /* initialized when started in `start` */
        private AtomicLong requested;
        private PriorityQueue<ValueSourcePair> queue; // not synchronized as
        // accessed inside
        // AtomicLong block
        private Set<InnerSubscriber> observers; // not synchronized as accessed
        // inside AtomicLong block

        public SortedMerge(final Comparator<T> comparator, final Subscriber<? super T> child) {
            this.comparator = comparator;
            this.child = child;
            child.add(childSubscription);
        }

        public void start(Iterable<Observable<? extends T>> os, AtomicLong requested) {
            this.requested = requested;

            // create a subscriber for each observable in the passed collection
            Map<InnerSubscriber, Observable<? extends T>> map = new HashMap<InnerSubscriber, Observable<? extends T>>();
            for (Observable<? extends T> o : os) {
                InnerSubscriber io = new InnerSubscriber();
                map.put(io, o);
                childSubscription.add(io);
            }

            // queue of values with their originating observers ordered by values
            queue = new PriorityQueue<ValueSourcePair>(map.size());

            // set of pending source observers - items will not be emitted down the
            // stream until the set is empty
            this.observers = new HashSet<InnerSubscriber>(map.keySet());

            // subscribe to all sources
            for (Map.Entry<InnerSubscriber, Observable<? extends T>> e : map.entrySet()) {
                e.getValue().unsafeSubscribe(e.getKey());
            }

        }

        /**
         * Tries to pick items from pending sources. Not synchronized as accessed
         * inside COUNTER_UPDATER block
         *
         * @param pending set of source observers pending a value for the merge
         * @return true if we need a shutdown
         */
        boolean pickPendingItems(Set<InnerSubscriber> pending) {
            // peek for a potential onCompleted event
            for (Iterator<InnerSubscriber> iter = pending.iterator(); iter.hasNext(); ) {
                InnerSubscriber subscr = iter.next();
                if (subscr.isUnsubscribed()) {
                    // we should remove completed observer from the pending set
                    iter.remove();
                }

                RxRingBuffer buffer = subscr.items;
                Object n = buffer.peek();

                if (n == null) {
                    // we should keep this observer as pending
                    continue;
                }

                if (buffer.isCompleted(n)) {
                    iter.remove();
                    // this observer indicates completion - unsubscribe and remove from
                    // pending
                    childSubscription.remove(subscr);
                    if (!childSubscription.hasSubscriptions()) {
                        // it is last upstream observer so onComplete so shut down
                        child.onCompleted();
                        return true;
                    }
                } else {
                    // got new data add to the priority queue and remove from pending set
                    @SuppressWarnings("unchecked")
                    T value = (T) buffer.getValue(n);
                    ValueSourcePair tuple = new ValueSourcePair(value, subscr);
                    queue.add(tuple);
                    // remove from pending set
                    iter.remove();
                }
            }
            return false;

        }

        /**
         * Polls a top item from the priority queue. Adds the top source back to
         * pending set. Not synchronized as accessed inside COUNTER_UPDATER block
         *
         * @param pending set of source observers pending a value for the merge
         * @return true if we need a shutdown
         */
        private boolean pollTopItem(Set<InnerSubscriber> pending) {
            // get a next item in order from the priority queue
            ValueSourcePair tuple = queue.poll();
            T value = tuple.value;

            InnerSubscriber observer = tuple.source;
            // mark the wining observer as pending again
            pending.add(observer);
            try {
                // emit the next item in order
                child.onNext(value);
                // we emitted so decrement the requested counter
                requested.decrementAndGet();
                observer.emitted++;
            } catch (Throwable e) {
                Exceptions.throwOrReport(e, child, value);
                return true;
            }
            // now remove
            RxRingBuffer buffer = observer.items;
            buffer.poll();
            // eagerly check if the next item on this queue is an onComplete
            if (buffer.isCompleted(buffer.peek())) {
                // we need to unsubscribe and remove from pending
                pending.remove(observer);
                childSubscription.remove(observer);
                if (!childSubscription.hasSubscriptions()) {
                    // it is last upstream observer so onComplete so shut down
                    child.onCompleted();
                    return true;
                }
                return false; // dont request for this observer
            }
            if (observer.emitted > THRESHOLD) {
                observer.requestMore(observer.emitted);
                observer.emitted = 0;
            }
            return false;
        }

        /**
         * check if we have values for each active (subscribed) source and emit top
         * value if we do
         *

         * This will only allow one thread at a time to do the work, but ensures via
         * `counter` increment/decrement that there is always once who acts on each
         * `tick`. Same concept as used in OperationObserveOn.
         */
        void tick() {
            final Set<InnerSubscriber> pending = this.observers;
            if (pending == null) {
                // nothing yet to do (initial request from Producer)
                return;
            }
            if (getAndIncrement() == 0) {
                final AtomicLong requested = this.requested;
                do {
                    while (true) {
                        if (pickPendingItems(pending)) {
                            return;
                        }
                        // we only emit if requested > 0 and no observer pending a value
                        if (requested.get() > 0 && pending.isEmpty()) {
                            if (pollTopItem(pending)) {
                                return;
                            }
                        } else {
                            break;
                        }
                    }
                } while (decrementAndGet() > 0);
            }

        }

        // used to observe each Observable we are sort-merging together
        // it collects one item per source into an internal priority queue
        final class InnerSubscriber extends Subscriber<T> {
            // Concurrent* since we need to read it from across threads
            final RxRingBuffer items = RxRingBuffer.getSpmcInstance();
            int emitted = 0; // not volatile/synchronized as accessed inside
            // COUNTER_UPDATER block

            @Override
            public void onStart() {
                request(RxRingBuffer.SIZE);
            }

            public void requestMore(long n) {
                request(n);
            }

            @Override
            public void onCompleted() {
                items.onCompleted();
                tick();
            }

            @Override
            public void onError(Throwable e) {
                // emit error immediately and shut down
                child.onError(e);
            }

            @Override
            public void onNext(Object t) {
                try {
                    items.onNext(t);
                } catch (MissingBackpressureException e) {
                    onError(e);
                }
                tick();
            }
        }

        ;

        final class ValueSourcePair implements Comparable<ValueSourcePair> {
            T value;
            InnerSubscriber source;

            ValueSourcePair(T value, InnerSubscriber source) {
                this.value = value;
                this.source = source;
            }

            @SuppressWarnings("unchecked")
            @Override
            public int compareTo(ValueSourcePair o) {
                if (comparator != null) {
                    return comparator.compare(value, o.value);
                } else {
                    Comparable<? super T> comparable = (Comparable<? super T>) value;
                    return comparable.compareTo(o.value);
                }
            }

            @Override
            public String toString() {
                return String.valueOf(value);
            }

        }

    }

    private final class MergeSortedSubscriber extends Subscriber<Observable<? extends T>> {

        final Subscriber<? super T> child;
        final SortedMerge<T> merge;
        final MergeSortedProducer<T> producer;
        final List<Observable<? extends T>> sources = new ArrayList<Observable<? extends T>>();

        public MergeSortedSubscriber(Subscriber<? super T> child, SortedMerge<T> merge, MergeSortedProducer<T> producer) {
            this.child = child;
            this.merge = merge;
            this.producer = producer;
        }

        @Override
        public void onCompleted() {
            if (sources.isEmpty()) {
                // this means we have not received a valid onNext before termination so
                // we emit the onCompleted
                child.onCompleted();
            } else {
                merge.start(sources, producer);
            }
        }

        @Override
        public void onError(Throwable e) {
            child.onError(e);
        }

        @Override
        public void onNext(Observable<? extends T> observable) {
            sources.add(observable);
        }

    }

}


text:
Capybara.app_host = "http://localhost:3000"
Capybara.default_driver = :selenium


text:
<?php
namespace Craft;

class ElementHelper
{
	// Public Methods
	// =========================================================================

	/**
	 * Sets a valid slug on a given element.
	 *
	 * @param BaseElementModel $element
	 *
	 * @return null
	 */
	public static function setValidSlug(BaseElementModel $element)
	{
		$slug = $element->slug;

		if (!$slug)
		{
			// Create a slug for them, based on the element's title.
			// Replace periods, underscores, and hyphens with spaces so they get separated with the slugWordSeparator
			// to mimic the default JavaScript-based slug generation.
			$slug = str_replace(array('.', '_', '-'), ' ', $element->title);

			// Enforce the limitAutoSlugsToAscii config setting
			if (craft()->config->get('limitAutoSlugsToAscii'))
			{
				if (!craft()->config->get('allowUppercaseInSlug'))
				{
					// Do this now because our ASCII character mappings are lowercase only.
					$slug = mb_strtolower($slug);
				}

				$slug = StringHelper::asciiString($slug);
			}
		}

		$element->slug = static::createSlug($slug);
	}

	/**
	 * Creates a slug based on a given string.
	 *
	 * @param string $str
	 *
	 * @return string
	 */
	public static function createSlug($str)
	{
		// Remove HTML tags
		$str = StringHelper::stripHtml($str);

		// Convert to kebab case
		$glue = craft()->config->get('slugWordSeparator');
		$lower = !craft()->config->get('allowUppercaseInSlug');
		$str = StringHelper::toKebabCase($str, $glue, $lower, false);

		return $str;
	}

	/**
	 * Sets the URI on an element using a given URL format, tweaking its slug if necessary to ensure it's unique.
	 *
	 * @param BaseElementModel $element
	 *
	 * @throws Exception
	 */
	public static function setUniqueUri(BaseElementModel $element)
	{
		$urlFormat = $element->getUrlFormat();

		// No URL format, no URI.
		if (!$urlFormat)
		{
			$element->uri  = null;
			return;
		}

		// No slug, or a URL format with no {slug}, just parse the URL format and get on with our lives
		if (!$element->slug || !static::doesUrlFormatHaveSlugTag($urlFormat))
		{
			$element->uri = craft()->templates->renderObjectTemplate($urlFormat, $element);
			return;
		}

		$uniqueUriConditions = array('and',
			'locale = :locale',
			'uri = :uri'
		);

		$uniqueUriParams = array(
			':locale' => $element->locale
		);

		if ($element->id)
		{
			$uniqueUriConditions[] = 'elementId != :elementId';
			$uniqueUriParams[':elementId'] = $element->id;
		}

		$slugWordSeparator = craft()->config->get('slugWordSeparator');
		$maxSlugIncrement = craft()->config->get('maxSlugIncrement');

		for ($i = 0; $i < $maxSlugIncrement; $i++)
		{
			$testSlug = $element->slug;

			if ($i > 0)
			{
				$testSlug .= $slugWordSeparator.$i;
			}

			$originalSlug = $element->slug;
			$element->slug = $testSlug;

			$testUri = craft()->templates->renderObjectTemplate($urlFormat, $element);

			// Make sure we're not over our max length.
			if (strlen($testUri) > 255)
			{
				// See how much over we are.
				$overage = strlen($testUri) - 255;

				// Do we have anything left to chop off?
				if (strlen($overage) > strlen($element->slug) - strlen($slugWordSeparator.$i))
				{
					// Chop off the overage amount from the slug
					$testSlug = $element->slug;
					$testSlug = substr($testSlug, 0, strlen($testSlug) - $overage);

					// Update the slug
					$element->slug = $testSlug;

					// Let's try this again.
					$i -= 1;
					continue;
				}
				else
				{
					// We're screwed, blow things up.
					throw new Exception(Craft::t('The maximum length of a URI is 255 characters.'));
				}
			}

			$uniqueUriParams[':uri'] = $testUri;

			$totalElements = craft()->db->createCommand()
				->select('count(id)')
				->from('elements_i18n')
				->where($uniqueUriConditions, $uniqueUriParams)
				->queryScalar();

			if ($totalElements ==  0)
			{
				// OMG!
				$element->slug = $testSlug;
				$element->uri = $testUri;
				return;
			}
			else
			{
				$element->slug = $originalSlug;
			}
		}

		throw new Exception(Craft::t('Could not find a unique URI for this element.'));
	}

	/**
	 * Returns whether a given URL format has a proper {slug} tag.
	 *
	 * @param string $urlFormat
	 *
	 * @return bool
	 */
	public static function doesUrlFormatHaveSlugTag($urlFormat)
	{
		$element = (object) array('slug' => StringHelper::randomString());
		$uri = craft()->templates->renderObjectTemplate($urlFormat, $element);

		return (strpos($uri, $element->slug) !== false);
	}

	/**
	 * Returns whether the given element is editable by the current user, taking user locale permissions into account.
	 *
	 * @param BaseElementModel $element
	 *
	 * @return bool
	 */
	public static function isElementEditable(BaseElementModel $element)
	{
		if ($element->isEditable())
		{
			if (craft()->isLocalized())
			{
				foreach ($element->getLocales() as $localeId => $localeInfo)
				{
					if (is_numeric($localeId) && is_string($localeInfo))
					{
						$localeId = $localeInfo;
					}

					if (craft()->userSession->checkPermission('editLocale:'.$localeId))
					{
						return true;
					}
				}
			}
			else
			{
				return true;
			}
		}

		return false;
	}

	/**
	 * Returns the editable locale IDs for a given element, taking user locale permissions into account.
	 *
	 * @param BaseElementModel $element
	 *
	 * @return array
	 */
	public static function getEditableLocaleIdsForElement(BaseElementModel $element)
	{
		$localeIds = array();

		if ($element->isEditable())
		{
			if (craft()->isLocalized())
			{
				foreach ($element->getLocales() as $localeId => $localeInfo)
				{
					if (is_numeric($localeId) && is_string($localeInfo))
					{
						$localeId = $localeInfo;
					}

					if (craft()->userSession->checkPermission('editLocale:'.$localeId))
					{
						$localeIds[] = $localeId;
					}
				}
			}
			else
			{
				$localeIds[] = craft()->i18n->getPrimarySiteLocaleId();
			}
		}

		return $localeIds;
	}

	/**
	 * Given an array of elements, will go through and set the appropriate "next"
	 * and "prev" elements on them.
	 *
	 * @param BaseElementModel[] $elements The array of elements.
	 */
	public static function setNextPrevOnElements($elements)
	{
		/** @var BaseElementModel $lastElement */
		$lastElement = null;

		foreach ($elements as $i => $element)
		{
			if ($lastElement)
			{
				$lastElement->setNext($element);
				$element->setPrev($lastElement);
			}
			else
			{
				$element->setPrev(false);
			}

			$lastElement = $element;
		}

		if ($lastElement)
		{
			$lastElement->setNext(false);
		}
	}
}


text:
using namespace System;
public ref class MarshalHelpers
{
public:
	static String^ ConvertString(std::string s);
};


text:
#region Apache License Version 2.0

#endregion Apache License Version 2.0

#if NET35 || NET40 || NET45 || NET461
using System.Web;
#else
using Microsoft.AspNetCore.Http;
#endif

namespace Senparc.Weixin.MP.TenPayLib
{
    public class RefundQueryRequestHandler : ClientRequestHandler
    {
        /// <summary>
        /// 退款明细查询接口
        /// </summary>
        /// <param name="httpContext"></param>
        public RefundQueryRequestHandler(HttpContext httpContext)
            : base(httpContext)
        {
            this.SetGateUrl("https://gw.tenpay.com/gateway/normalrefundquery.xml");
        }
    }
}


text:
<!DOCTYPE html>

		<title>Demo Express Web Application</title>

Form Submission Successful!

		<form name="contact-form">

				Name: <input type="text" value="Bob" name="contact-name">

				Age: <input type="number" value="45" name="contact-age">

				Phone: <input type="tel" value="111-222-3333" name="contact-phone">

				Email: <input type="email" value="bob@somedomain.bob" name="contact-email">

			<button type="button">Send</button>

			<select name='post-method'>
				<option value='url-encoded-post'>POST URL Encoded Data</option>
				<option value='json-post'>POST JSON Data</option>
			</select>

		</form>


text:
ENV['RAILS_ENV'] ||= 'test'
SPEC_ROOT  = Pathname(File.dirname(__FILE__))
DUMMY_ROOT = SPEC_ROOT.join('../dummy')

require 'spec_helper/codeclimate' # this should be atop all
require 'spec_helper/dummy_and_test_dependencies'
require 'spec_helper/support'
require 'spec_helper/create_database'
require 'spec_helper/configure_rspec'


text:
package com.goree.api.domain;

import lombok.Getter;
import lombok.Setter;
import lombok.ToString;

import java.util.List;

@Getter
@Setter
@ToString
public class Group {
    private long id;
    private Member leader;
    private String name;
    private String description;
    private List<Member> members;
    private long memberCount;
    private String imagePath;

    public boolean equals(Object other){
        if(!(other instanceof Group))
            return false;
        Group group = (Group)other;

        return
                group.getLeader() == null ? getLeader() == null : group.getLeader().getId() == getLeader().getId()
             && group.getName().equals(getName());

    }
}


text:
using namespace System::Collections::Generic;
using namespace System::Collections::Specialized;
using namespace System::Security::Cryptography::X509Certificates;
using namespace CefSharp::Internals::Serialization;

namespace CefSharp
{
    namespace Internals
    {
        //Static class for converting to/from Cef classes
        private ref class TypeConversion abstract sealed
        {
        public:
            //Convert from NameValueCollection to HeaderMap
            static CefResponse::HeaderMap ToNative(NameValueCollection^ headers)
            {
                CefResponse::HeaderMap result;

                if (headers == nullptr)
                {
                    return result;
                }

                for each (String^ key in headers)
                {
                    for each(String^ value in headers->GetValues(key))
                    {
                        result.insert(std::pair<CefString, CefString>(StringUtils::ToNative(key), StringUtils::ToNative(value)));
                    }
                }

                return result;
            }

            //ConvertFrom CefDownload to DownloadItem
            static DownloadItem^ FromNative(CefRefPtr<CefDownloadItem> downloadItem)
            {
                auto item = gcnew CefSharp::DownloadItem();
                item->IsValid = downloadItem->IsValid();
                //NOTE: Description for IsValid says `Do not call any other methods if this function returns false.` so only load if IsValid = true
                if (item->IsValid)
                {
                    item->IsInProgress = downloadItem->IsInProgress();
                    item->IsComplete = downloadItem->IsComplete();
                    item->IsCancelled = downloadItem->IsCanceled();
                    item->CurrentSpeed = downloadItem->GetCurrentSpeed();
                    item->PercentComplete = downloadItem->GetPercentComplete();
                    item->TotalBytes = downloadItem->GetTotalBytes();
                    item->ReceivedBytes = downloadItem->GetReceivedBytes();
                    item->StartTime = FromNative(downloadItem->GetStartTime());
                    item->EndTime = FromNative(downloadItem->GetEndTime());
                    item->FullPath = StringUtils::ToClr(downloadItem->GetFullPath());
                    item->Id = downloadItem->GetId();
                    item->Url = StringUtils::ToClr(downloadItem->GetURL());
                    item->OriginalUrl = StringUtils::ToClr(downloadItem->GetOriginalUrl());
                    item->SuggestedFileName = StringUtils::ToClr(downloadItem->GetSuggestedFileName());
                    item->ContentDisposition = StringUtils::ToClr(downloadItem->GetContentDisposition());
                    item->MimeType = StringUtils::ToClr(downloadItem->GetMimeType());
                }

                return item;
            }

            //Convert from CefTime to Nullable<DateTime>
            static Nullable<DateTime> FromNative(CefTime time)
            {
                auto epoch = time.GetDoubleT();
                if (epoch == 0)
                {
                    return Nullable<DateTime>();
                }
                return Nullable<DateTime>(DateTime(1970, 1, 1, 0, 0, 0).AddSeconds(epoch).ToLocalTime());
            }

            static WebPluginInfo^ FromNative(CefRefPtr<CefWebPluginInfo> webPluginInfo)
            {
                return gcnew WebPluginInfo(StringUtils::ToClr(webPluginInfo->GetName()),
                    StringUtils::ToClr(webPluginInfo->GetDescription()),
                    StringUtils::ToClr(webPluginInfo->GetPath()),
                    StringUtils::ToClr(webPluginInfo->GetVersion()));
            }

            static IList<DraggableRegion>^ FromNative(const std::vector<CefDraggableRegion>& regions)
            {
                if (regions.size() == 0)
                {
                    return nullptr;
                }

                auto list = gcnew List<DraggableRegion>();

                for each (CefDraggableRegion region in regions)
                {
                    list->Add(DraggableRegion(region.bounds.width, region.bounds.height, region.bounds.x, region.bounds.y, region.draggable == 1));
                }

                return list;
            }

            static CefRefPtr<CefValue> ToNative(Object^ value)
            {
                auto cefValue = CefValue::Create();

                if (value == nullptr)
                {
                    cefValue->SetNull();

                    return cefValue;
                }

                auto type = value->GetType();
                Type^ underlyingType = Nullable::GetUnderlyingType(type);
                if (underlyingType != nullptr)
                {
                    type = underlyingType;
                }

                if (type == Boolean::typeid)
                {
                    cefValue->SetBool(safe_cast<bool>(value));
                }
                else if (type == Int32::typeid)
                {
                    cefValue->SetInt(safe_cast<int>(value));
                }
                else if (type == String::typeid)
                {
                    cefValue->SetString(StringUtils::ToNative(safe_cast<String^>(value)));
                }
                else if (type == Double::typeid)
                {
                    cefValue->SetDouble(safe_cast<double>(value));
                }
                else if (type == Decimal::typeid)
                {
                    cefValue->SetDouble(Convert::ToDouble(value));
                }
                else if (System::Collections::IDictionary::typeid->IsAssignableFrom(type))
                {
                    auto dictionary = (System::Collections::IDictionary^) value;
                    auto cefDictionary = CefDictionaryValue::Create();

                    for each (System::Collections::DictionaryEntry entry in dictionary)
                    {
                        auto key = StringUtils::ToNative(Convert::ToString(entry.Key));
                        auto entryValue = entry.Value;
                        //We don't pass a nameConverter here as the keys should
                        //remain unchanged
                        SerializeV8Object(cefDictionary, key, entryValue, nullptr);
                    }

                    cefValue->SetDictionary(cefDictionary);
                }
                else if (System::Collections::IEnumerable::typeid->IsAssignableFrom(type))
                {
                    auto enumerable = (System::Collections::IEnumerable^) value;
                    auto cefList = CefListValue::Create();

                    int i = 0;
                    for each (Object^ arrObj in enumerable)
                    {
                        //We don't pass a nameConverter here as the keys should
                        //remain unchanged
                        SerializeV8Object(cefList, i, arrObj, nullptr);

                        i++;
                    }
                    cefValue->SetList(cefList);
                }

                return cefValue;
            }

            static Object^ FromNative(const CefRefPtr<CefValue>& value)
            {
                if (!value.get())
                {
                    return nullptr;
                }

                auto type = value->GetType();

                if (type == CefValueType::VTYPE_BOOL)
                {
                    return value->GetBool();
                }

                if (type == CefValueType::VTYPE_DOUBLE)
                {
                    return value->GetDouble();
                }

                if (type == CefValueType::VTYPE_INT)
                {
                    return value->GetInt();
                }

                if (type == CefValueType::VTYPE_STRING)
                {
                    return StringUtils::ToClr(value->GetString());
                }

                if (type == CefValueType::VTYPE_DICTIONARY)
                {
                    return FromNative(value->GetDictionary());
                }

                if (type == CefValueType::VTYPE_LIST)
                {
                    return FromNative(value->GetList());
                }

                return nullptr;
            }

            static IDictionary<String^, Object^>^ FromNative(const CefRefPtr<CefDictionaryValue>& dictionary)
            {
                if (!dictionary.get() || dictionary->GetSize() == 0)
                {
                    return nullptr;
                }

                auto dict = gcnew Dictionary<String^, Object^>();

                CefDictionaryValue::KeyList keys;
                dictionary->GetKeys(keys);

                for (size_t i = 0; i < keys.size(); i++)
                {
                    auto key = StringUtils::ToClr(keys[i]);
                    auto value = DeserializeObject(dictionary, keys[i], nullptr);

                    dict->Add(key, value);
                }

                return dict;
            }

            static List<Object^>^ FromNative(const CefRefPtr<CefListValue>& list)
            {
                auto result = gcnew List<Object^>(list->GetSize());
                for (size_t i = 0; i < list->GetSize(); i++)
                {
                    result->Add(DeserializeObject(list, i, nullptr));
                }

                return result;
            }

            // Copied from CefSharp.BrowserSubprocess.Core\TypeUtils.h since it can't be included
            static DateTime ConvertCefTimeToDateTime(CefTime time)
            {
                return DateTimeUtils::FromCefTime(time.year,
                    time.month,
                    time.day_of_month,
                    time.hour,
                    time.minute,
                    time.second,
                    time.millisecond);
            }

            static Cookie^ FromNative(const CefCookie& cefCookie)
            {
                auto cookie = gcnew Cookie();
                auto cookieName = StringUtils::ToClr(cefCookie.name);

                if (!String::IsNullOrEmpty(cookieName))
                {
                    cookie->Name = cookieName;
                    cookie->Value = StringUtils::ToClr(cefCookie.value);
                    cookie->Domain = StringUtils::ToClr(cefCookie.domain);
                    cookie->Path = StringUtils::ToClr(cefCookie.path);
                    cookie->Secure = cefCookie.secure == 1;
                    cookie->HttpOnly = cefCookie.httponly == 1;
                    cookie->Creation = ConvertCefTimeToDateTime(cefCookie.creation);
                    cookie->LastAccess = ConvertCefTimeToDateTime(cefCookie.last_access);
                    cookie->SameSite = (CefSharp::Enums::CookieSameSite)cefCookie.same_site;
                    cookie->Priority = (CefSharp::Enums::CookiePriority)cefCookie.priority;

                    if (cefCookie.has_expires)
                    {
                        cookie->Expires = ConvertCefTimeToDateTime(cefCookie.expires);
                    }
                }

                return cookie;
            }

            static NavigationEntry^ FromNative(const CefRefPtr<CefNavigationEntry> entry, bool current)
            {
                SslStatus^ sslStatus;

                if (!entry.get())
                {
                    return nullptr;
                }

                if (!entry->IsValid())
                {
                    return gcnew NavigationEntry(current, DateTime::MinValue, nullptr, -1, nullptr, nullptr, (TransitionType)-1, nullptr, false, false, sslStatus);
                }

                auto time = entry->GetCompletionTime();
                DateTime completionTime = CefTimeUtils::ConvertCefTimeToDateTime(time.GetDoubleT());
                auto ssl = entry->GetSSLStatus();
                X509Certificate2^ sslCertificate;

                if (ssl.get())
                {
                    auto certificate = ssl->GetX509Certificate();
                    if (certificate.get())
                    {
                        auto derEncodedCertificate = certificate->GetDEREncoded();
                        auto byteCount = derEncodedCertificate->GetSize();
                        if (byteCount > 0)
                        {
                            auto bytes = gcnew cli::array<Byte>(byteCount);
                            pin_ptr<Byte> src = &bytes[0]; // pin pointer to first element in arr

                            derEncodedCertificate->GetData(static_cast<void*>(src), byteCount, 0);

                            sslCertificate = gcnew X509Certificate2(bytes);
                        }
                    }

                    sslStatus = gcnew SslStatus(ssl->IsSecureConnection(), (CertStatus)ssl->GetCertStatus(), (SslVersion)ssl->GetSSLVersion(), (SslContentStatus)ssl->GetContentStatus(), sslCertificate);
                }

                return gcnew NavigationEntry(current, completionTime, StringUtils::ToClr(entry->GetDisplayURL()), entry->GetHttpStatusCode(),
                    StringUtils::ToClr(entry->GetOriginalURL()), StringUtils::ToClr(entry->GetTitle()), (TransitionType)entry->GetTransitionType(),
                    StringUtils::ToClr(entry->GetURL()), entry->HasPostData(), true, sslStatus);
            }
        };
    }
}