WALTHAM, Mass., Jun 08, 2010 (BUSINESS WIRE) --ModusLink Global Solutions(TM), Inc. (NASDAQ: MLNK) today reported financial results for its third quarter of fiscal year 2010 ended April 30, 2010.
"Revenue for the third quarter of fiscal 2010 was less than the same quarter last year, primarily due to lower revenue from new client engagements. As we work to build our revenue from new engagements back toward the record levels achieved in fiscal 2009, the time required to start new programs and bring them to the point of generating significant revenue continues to be longer than what we experienced previously," said Joseph C. Lawler, chairman, president and chief executive officer of ModusLink Global Solutions. "However, we are encouraged with the signs of stabilization in our business, including higher unit volumes from certain client programs and revenue from our base business that is comparable to the same quarter last year."
"Gross margin for the third quarter was lower than the same period last year, primarily due to lower revenue and higher costs related to the start-up of new engagements. For the first nine months of this year, gross margin was 13%, which is within our previously stated target range and represents good margin expansion compared with the same period last year," continued Lawler.
"Looking forward, we continue to have a very compelling value proposition for outsourced supply chain, aftermarket and e-Business solutions and a high level of client satisfaction. Our pipeline of opportunities for new programs remains strong and is positioned to make a greater contribution to our revenue. In the near-term, we are focused on adding new client programs from existing and prospective clients, implementing our plans to bring new programs to the point of revenue generation faster and delivering the highest level of client service," continued Lawler.
ModusLink reported net revenue of $213.7 million for the third quarter of fiscal 2010, a decrease of 7.7% compared to net revenue of $231.5 million reported in the third quarter of fiscal 2009. Net revenue for the third quarter of 2010 reflects lower revenue from new client engagements compared with the third quarter of 2009, which was a record year for revenue from new engagements. Revenue for the third quarter of 2010 included $8.9 million from Tech for Less, which was acquired on December 4, 2009.
Gross profit for the third quarter of fiscal 2010 was $24.6 million, or 11.5% of revenue, compared to $32.4 million, or 14.0% of revenue, in the third quarter of fiscal 2009. The decrease in gross margin as a percentage of revenue in the third quarter of fiscal 2010 was primarily due to lower revenue and higher costs related to the start-up of new engagements compared to the same period last year.
Operating income for the third quarter of fiscal 2010 was $1.3 million compared to operating income of $2.7 million in the third quarter of fiscal 2009. The decline in operating income was principally due to lower revenue and gross margin in the third quarter of fiscal 2010 compared to the third quarter of the prior year.
Net loss for the third quarter of 2010 was $3.4 million, or ($0.08) per share, compared to net loss of $1.9 million, or ($0.04) per share, for the same period in fiscal 2009. Third quarter fiscal 2010 results included a loss from discontinued operations of $2.3 million, or ($0.05) per share, primarily due to updated sublease assumptions for a facility no longer being utilized for operations by the Company, compared to a loss from discontinued operations of $68 thousand, or ($0.00) per share, in fiscal 2009.
Excluding net charges related to depreciation, amortization of intangibles, stock-based compensation, restructuring and non-recurring charges, the Company reported non-GAAP operating income of $8.3 million for the third quarter of fiscal 2010, a 48.5% decrease from $16.1 million for the same period in fiscal 2009.
As of April 30, 2010, the Company had working capital of approximately $221.3 million compared to $237.0 million at July 31, 2009 and $236.7 million at April 30, 2009. Included in working capital as of April 30, 2010 were cash, cash equivalents, short-term investments and marketable securities totaling $160.2 million compared to $179.2 million at July 31, 2009 and $167.8 million at April 30, 2009. The Company concluded the quarter with no outstanding bank debt.
For the third quarter of fiscal 2010, net cash provided by operating activities of continuing operations was $3.7 million and additions to property and equipment were $3.1 million, and therefore free cash flow from operations was $535 thousand, compared to $16.0 million in the same period in 2009.
"We continue to effectively manage costs and maintain working capital efficiency," said Steven G. Crane, chief financial officer of ModusLink Global Solutions. "ModusLink has a very strong and liquid balance sheet, providing us with competitive advantage in our market and a platform for future acquisitions and long-term growth."
Based on current forecasts from its clients and expected contributions from new client engagements, ModusLink continues to expect a sequential improvement in revenue for the fourth quarter of fiscal 2010, compared to the third quarter of fiscal 2010. In addition, ModusLink continues to expect to generate positive free cash flow from operations for the full fiscal year 2010 and at a level similar to fiscal year 2009.
The Company also announced that its Board of Directors has authorized the repurchase of up to $10 million of the Company's common stock from time to time over the next 18 months and that it has completed its previously announced repurchase program.
During the third quarter of fiscal 2010, the Company completed the stock repurchase program that was announced on June 9, 2009 and pursuant to which it was authorized to repurchase $15 million of common stock. In the third quarter, the Company repurchased 232,865 shares of its common stock for aggregate consideration of $2.2 million, bringing the total number of shares repurchased under the program to 1,805,105.
As previously announced, ModusLink Global Solutions, Inc. will hold a conference call to discuss its fiscal 2010 third quarter results at 5:00 p.m. ET on June 8, 2010. Investors can listen to the conference call on the Internet at http://www.ir.moduslink.com. To listen to the live call, go to the website at least 15 minutes prior to the start time to download and install the necessary audio software.
ModusLink Global Solutions, Inc. (NASDAQ: MLNK) designs and executes global supply chain, aftermarket and e-Business solutions -- across multiple channels -- for the world's leading technology and consumer brands. More than 25 years of experience developing business process outsourcing and technology solutions for the global value chain enables the Company to solve clients' cost, time-to-market, customer satisfaction and revenue objectives. ModusLink Global Solutions has headquarters in Waltham, Massachusetts and more than 25 facilities in 14 countries - giving it the largest global footprint in the industry. For additional information, visit http://www.moduslink.com or visit ModusLink's value chain blog, Value Unchained at http://www.valueunchained.com.
The Company believes that its non-GAAP measure of operating income/(loss) ("non-GAAP operating income/(loss)") provides investors with a useful, supplemental measure of the Company's operating performance by excluding the impact of non-cash charges and restructuring activities. Each of the excluded items was excluded because it may be considered to be of a non-operational or non-cash nature. Historically, the Company has recorded significant impairment and restructuring charges. These charges, as well as charges related to depreciation, amortization of intangible assets and stock-based compensation, have been excluded for the purpose of enhancing the understanding by both management and investors of the underlying baseline operating results and trends of the business, which management uses to evaluate our financial performance for purposes of planning and forecasting future periods. Non-GAAP operating income/(loss) does not have any standardized definition and, therefore, is unlikely to be comparable to similar measures presented by other reporting companies. Non-GAAP operating income/(loss) should not be evaluated in isolation of, or as a substitute for, the Company's financial results prepared in accordance with United States generally accepted accounting principles. The Company's usage of non-GAAP operating income/(loss), and the underlying methodology in excluding certain charges, is not necessarily an indication of the results of operations that may be expected in the future, or that the Company will not, in fact, incur such charges in future periods. A table reconciling the Company's non-GAAP operating income/(loss) to its GAAP operating income/(loss) and its GAAP net income/(loss) is included in the statement of operations information in this release. The Company defines "Free cash flow from operations" as net cash provided by (used in) operating activities of continuing operations less additions to property and equipment as shown on the Company's consolidated statements of cash flows. Management considers free cash flow from operations to be an important indicator of the Company's financial strength and the ability to generate liquidity because it reflects net cash generated from operations. Free cash flow from operations is not a measure determined in accordance with GAAP and may not be defined and calculated by other companies in the same manner. A table reconciling "Free cash flow from operations" to the GAAP measures of net cash provided by (used in) operating activities of continuing operations less additions to property and equipment is included in the statement of operations information in this release.
This release contains forward-looking statements, which address a variety of subjects including, for example, the Company's pipeline of opportunities for new programs and their impact on revenue, near-term initiatives, the prospects for future growth and strategic acquisitions and the current expectations regarding revenue in the fourth quarter of fiscal 2010 and free cash flow from operations for fiscal 2010.All statements other than statements of historical fact, including without limitation, those with respect to the Company's goals, plans, expectations and strategies set forth herein are forward-looking statements. The following important factors and uncertainties, among others, could cause actual results to differ materially from those described in these forward-looking statements: the Company's success, including its ability to meet its revenue and operating income targets, maintain and improve its free cash flow from operations and cash position, expand its operations and revenue, lower its costs, improve its gross margins, sustain profitability, reach its long-term objectives and operate optimally, depends on its ability to execute on its business strategy and the continued and increased demand for and market acceptance of its services; global economic conditions, especially in the technology sector are uncertain and subject to volatility; demand for our clients' products may decline or may not achieve the levels anticipated by our clients; potential acquisitions may not be available on terms acceptable to the Company or at all; the Company's management may face strain on managerial and operational resources as they try to oversee the expanded operations; the Company may not realize the expected benefits of its restructuring and cost cutting actions; the Company may not be able to expand its operations in accordance with its business strategy; the Company's cash balances may not be sufficient to allow the Company to meet all of its business and investment goals; the Company may experience difficulties integrating technologies, operations and personnel in accordance with its business strategy; the Company derives a significant portion of its revenue from a small number of customers and the loss of any of those customers could significantly damage the Company's financial condition and results of operations; the Company frequently sells to its clients on a purchase order basis rather than pursuant to contracts with minimum purchase requirements, and therefore its sales and the amount of projected revenue that is actually realized are subject to demand variability; risks inherent with conducting international operations; tax rate expectations are based on current tax law and current expected income and may be affected by the jurisdictions in which profits are determined to be earned and taxed, changes in estimates of credits, benefits and deductions, the resolution of issues arising from tax audits with various tax authorities, including payment of interest and penalties and the ability to realize deferred tax assets;the mergers and acquisitions and IPO markets are inherently unpredictable and liquidity events for companies in the Company's venture capital portfolio may not occur; and increased competition and technological changes in the markets in which the Company competes.For a detailed discussion of cautionary statements that may affect the Company's future results of operations and financial results, please refer to the Company's filings with the Securities and Exchange Commission, including the Company's most recent Quarterly Reporton Form 10-Q and Annual Report on Form 10-K. Forward-looking statements represent management's current expectations and are inherently uncertain. We do not undertake any obligation to update forward-looking statements made by us.
Note: Non-GAAP operating income represents total operating income, excluding net charges related to depreciation, amortization of intangible assets, stock-based compensation, restructuring, and non-recurring charges.


No one counted noses but it is fair to estimate that millions joined in the "For Pete's Sake, Sing" celebrations of Pete Seeger's 90th birthday Sunday May 3. The birthday parties ranged from a sold-out concert at Madison Square Garden featuring Bruce Springsteen, other superstars and Pete Seeger himself, to local hootenannies in cities and towns all across the country and around the world.
Pete Seeger has reached his ninth decade of life still in the thick of the struggle for world peace, protection of the environment, racial equality and union rights for workers.
Pete Seeger sang to millions at Barack Obama's pre-inaugural celebration at the Lincoln Memorial. Isn't it time to give the highest possible honor to this man, the Nobel Peace Prize? So far, more than 21,000 poeople have signed petitions urging a Nobel for Seeger. Rep. Barbara Lee (D-Calif.), herself a tribune of world peace, has agreed to carry the petitions to Stockholm. This would not be simply an honor for Seeger as an individual. It would also honor the many millions of working people who stood along with Seeger creating a broad cultural movement that has played a huge role in opposing war, racism, the exploitation of human beings and the degradation of planet earth by giant banks and corporations.


the security and privacy of our Client's information.
Click here to view our Corporate Account Takeover and Information Security Awareness training.
Click here for a listing of financial scam alerts.
Click here for a listing of our security tips.
We are continually monitoring online threats and reviewing our security procedures to protect all Client information. Although the Internet is one of the most powerful communication tools available, criminals may use the Internet to defraud unsuspecting people. Scams to retrieve a Client's personal information can occur through many different channels, such as a phone call, email message or social engineering technique.
One of the most common methods used today to retrieve a Client's personal information is called "phishing." This is when fraudulent websites or emails are created in an attempt to steal personal information.
Clients are advised to never give out confidential information or passwords by replying to an email or by going to a website through a link included in an email. Even if you click on the link in an email but don't actually provide confidential information, you could be exposing yourself to viruses, malware or other harmful pieces of software. Remember, American Momentum Bank does not request confidential, personal or secure login information via email.
Spammers have gotten creative by making spam email messages appear as though it came from a reputable company or government agency, thus creating a sense of urgency to respond. This is a technique frequently used to lure unsuspecting people to provide confidential information that may be used for identity theft.
Be cautious of anyone calling you to ask for bank account or personal information over the phone.
Clients are urged to protect their computer system through the use of anti-virus, anti-spyware and firewall hardware and/or software. If anti-virus software is used, it's very important to keep the virus definitions up-to-date so that the most recent threats may be detected.
If you use an Operating System such as Microsoft® Windows XP or Vista, stay abreast of the many security updates Microsoft® releases. It is important that your computer is updated and contains the appropriate patches. You may choose to setup an automatic update at a certain time each day or week.
Review your account statements when they arrive and report discrepancies to American Momentum Bank.
If you do not recognize the sender of an email message, delete the email without opening it.
Keep your passwords confidential. Change passwords regularly using a complex combination of letters, numbers and special characters. Avoid using obvious passwords that may be easily guessed or hacked.
Never dispose of a hard drive without thoroughly cleaning it to remove all personal information.
When using an ATM or card machine at a gas pump, carefully inspect the device to ensure that no abnormal attachments have been added. If it doesn't look right, don't use it.
If you believe you are a victim of fraud or need to report a suspicious email that uses American Momentum Bank's name, please forward it to us immediately at: reportfraud@americanmomentum.bank. If you receive a suspicious phone call that uses American Momentum Bank's name, please contact the bank at (866) 530-2265.


For many years, I was happy and content dyeing with plants found practically under my feet. Sometimes, when studying dusty old dyeing books, I saw intimidating indigo recipes, which required purchase and use of chemicals I couldn't pronounce, and even less understand the effects and consequences of. So I didn't make blues, even though it's a favourite colour of mine. But didn't care that much, really, there's a big world of colours, besides blue.
After mixing the ingredients and placing the bucket (with lid) a warm place, all you have to do is gently stirring once a day.
Wow, I love indigo! I've always loved the colour and the name of it too. When you say 'washing soda' do you mean sodium bicarbonate or some other thing?
I love the sheep picture--it's like a before and after of white sheep wool, then some dyed indigo wool.
I have some frozen japanese indigo leaves and liquid left over from last summer.
I did not get a good blue from them, but a turquoise greenish colour, and not wanting to throw away the dye, I froze it.
Should I dare defrost and try again - any idea how what to expect?
Hi Joanne, what method did you use? And did you freeze fresh leaves, or soaked?
I used the cold water method - so i wasn't expecting indigo blues - and I froze soaked chopped up leaves...what do you think? I also dared to dry and crush fresh leaves. I don't have a lot three or four tablespoons, but how should I treat them for dye?
Considering the dried leaves, you haven't much to work with. Maybe you could try a gather some more this year, and extract the indigotin. Extracting requires a fair amount of leaves. Hope this can be of help to you.
I read your blog the fermentation of Indigo.
Do I have to mix the recipe you showed into the plane water? I can't image only that materials is going on fermentation.


A while back I was set a challenge by the creative writing tutor here. He's heavily into his surrealism and his comedy and he set me a task to write a piece along those lines. I'm not sure quite how comedic it turned out, but what I came up with was definitely surreal.
The theme he set me was 'dentists' and it took a while for me to think of anything to write about that at all, but once I got going, the story seemed to write itself. You can read what I came up with here.
I have to say, I do like a challenge when it comes to writing so if any of you ever want to set one, you can leave it as a comment or send it as a mail and I'll definitely try to rise to it. What am I letting myself in for?!
Write about a married woman who visits the hospital for an eye operation.
Great story but my chilled demeanour going to the Dentist has changed to abject terror (I have an appointment on Monday).
Your story put me in mind of *In With the Doctor*, a short story from James Kelman's *Greyhound for Breakfast*. He is a relatively obscure Glasgow writer.
Your tale also confirmed my assertion, which many find unbelievable, that HMP contains many talented people. For me and for my culture, storytelling is one of the finest things and you've certainly got it. Use it well.


text:
package dpbuffered

import (
	"sync"
	"sync/atomic"

	"context"
	"fmt"
	"github.com/signalfx/gateway/logkey"
	"github.com/signalfx/golib/datapoint"
	"github.com/signalfx/golib/datapoint/dpsink"
	"github.com/signalfx/golib/event"
	"github.com/signalfx/golib/log"
	"github.com/signalfx/golib/pointer"
	"github.com/signalfx/golib/sfxclient"
	"github.com/signalfx/golib/trace"
	"net/http"
	"runtime"
)

// Config controls BufferedForwarder limits
type Config struct {
	BufferSize         *int64
	MaxTotalDatapoints *int64
	MaxTotalEvents     *int64
	MaxTotalSpans      *int64
	MaxDrainSize       *int64
	NumDrainingThreads *int64
	Checker            *dpsink.ItemFlagger
	Cdim               *log.CtxDimensions
	Name               *string
	UseAuthFromRequest *bool
}

func (c *Config) String() string {
	return fmt.Sprintf("Config [Name: %s BufferSize: %d MaxTotalDatapoints: %d MaxTotalEvents: %d MaxTotalSpans: %d MaxDrainSize: %d NumDrainingThreads: %d UseAuthFromRequest: %t", *c.Name, *c.BufferSize, *c.MaxTotalDatapoints, *c.MaxTotalEvents, *c.MaxTotalSpans, *c.MaxDrainSize, *c.NumDrainingThreads, *c.UseAuthFromRequest)
}

// DefaultConfig are default values for buffered forwarders
var DefaultConfig = &Config{
	BufferSize:         pointer.Int64(1000000),
	MaxTotalDatapoints: pointer.Int64(1000000),
	MaxTotalEvents:     pointer.Int64(1000000),
	MaxTotalSpans:      pointer.Int64(1000000),
	MaxDrainSize:       pointer.Int64(30000),
	NumDrainingThreads: pointer.Int64(int64(runtime.NumCPU())),
	Name:               pointer.String(""),
	UseAuthFromRequest: pointer.Bool(false),
}

// Sink is a dpsink and trace.sink
type Sink interface {
	dpsink.Sink
	trace.Sink
}

type stats struct {
	totalDatapointsBuffered int64
	totalEventsBuffered     int64
	datapointsInFlight      int64
	eventsInFlight          int64
	totalTracesBuffered     int64
	tracesInFlight          int64
}

// BufferedForwarder abstracts out datapoint buffering.  Points put on its channel are buffered
// and sent in large groups to a waiting sink
type BufferedForwarder struct {
	dpChan                      chan []*datapoint.Datapoint
	eChan                       chan []*event.Event
	tChan                       chan []*trace.Span
	config                      *Config
	stats                       stats
	useAuthFromRequest          bool
	threadsWaitingToDie         sync.WaitGroup
	blockingDrainWaitMutex      sync.Mutex
	blockingEventDrainWaitMutex sync.Mutex
	blockingTraceDrainWaitMutex sync.Mutex
	// these are where we hold stuff that weren't the correct token
	// these are protected by the three mutexes above
	holdDatapoints []*datapoint.Datapoint
	holdEvents     []*event.Event
	holdSpans      []*trace.Span

	logger     log.Logger
	checker    *dpsink.ItemFlagger
	cdim       *log.CtxDimensions
	identifier string

	sendTo         Sink
	stopContext    context.Context
	stopFunc       context.CancelFunc
	closeSender    func() error
	afterStartup   func() error
	debugEndpoints func() map[string]http.Handler
}

// DebugEndpoints returns any configured http handlers and the point they can be reached at
func (forwarder *BufferedForwarder) DebugEndpoints() map[string]http.Handler {
	return forwarder.debugEndpoints()
}

var _ dpsink.Sink = &BufferedForwarder{}

type errDPBufferFull string

func (e errDPBufferFull) Error() string {
	return "Forwarder " + string(e) + " unable to send more datapoints.  Buffer full"
}

// AddDatapoints sends the datapoints to a chan buffer that eventually is flushed in big groups
func (forwarder *BufferedForwarder) AddDatapoints(ctx context.Context, points []*datapoint.Datapoint) error {
	if forwarder.checker.CtxFlagCheck.HasFlag(ctx) {
		forwarder.cdim.With(ctx, forwarder.logger).Log("Datapoint call recieved in buffered forwarder")
	}
	atomic.AddInt64(&forwarder.stats.totalDatapointsBuffered, int64(len(points)))
	if *forwarder.config.MaxTotalDatapoints <= atomic.LoadInt64(&forwarder.stats.totalDatapointsBuffered) {
		atomic.AddInt64(&forwarder.stats.totalDatapointsBuffered, int64(-len(points)))
		return errDPBufferFull(forwarder.identifier)
	}
	select {
	case forwarder.dpChan <- points:
		return nil
	case <-forwarder.stopContext.Done():
		return forwarder.stopContext.Err()
	case <-ctx.Done():
		return ctx.Err()
	}
}

type errEBufferFull string

func (e errEBufferFull) Error() string {
	return "Forwarder " + string(e) + " unable to send more events.  Buffer full"
}

// AddEvents sends the events to a chan buffer that eventually is flushed in big groups
func (forwarder *BufferedForwarder) AddEvents(ctx context.Context, events []*event.Event) error {
	if forwarder.checker.CtxFlagCheck.HasFlag(ctx) {
		forwarder.cdim.With(ctx, forwarder.logger).Log("Events call received in buffered forwarder")
	}
	atomic.AddInt64(&forwarder.stats.totalEventsBuffered, int64(len(events)))
	if *forwarder.config.MaxTotalEvents <= atomic.LoadInt64(&forwarder.stats.totalEventsBuffered) {
		atomic.AddInt64(&forwarder.stats.totalEventsBuffered, int64(-len(events)))
		return errEBufferFull(forwarder.identifier)
	}
	select {
	case forwarder.eChan <- events:
		return nil
	case <-forwarder.stopContext.Done():
		return forwarder.stopContext.Err()
	case <-ctx.Done():
		return ctx.Err()
	}
}

type errTBufferFull string

func (e errTBufferFull) Error() string {
	return "Forwarder " + string(e) + " unable to send more traces.  Buffer full"
}

// AddSpans sends the traces to a chan buffer that traceually is flushed in big groups
func (forwarder *BufferedForwarder) AddSpans(ctx context.Context, traces []*trace.Span) error {
	atomic.AddInt64(&forwarder.stats.totalTracesBuffered, int64(len(traces)))
	if *forwarder.config.MaxTotalSpans <= atomic.LoadInt64(&forwarder.stats.totalTracesBuffered) {
		atomic.AddInt64(&forwarder.stats.totalTracesBuffered, int64(-len(traces)))
		return errTBufferFull(forwarder.identifier)
	}
	select {
	case forwarder.tChan <- traces:
		return nil
	case <-forwarder.stopContext.Done():
		return forwarder.stopContext.Err()
	case <-ctx.Done():
		return ctx.Err()
	}
}

// DebugDatapoints returns debug level datapoints about this forwarder, including errors processing datapoints
func (forwarder *BufferedForwarder) DebugDatapoints() []*datapoint.Datapoint {
	return []*datapoint.Datapoint{
		sfxclient.Gauge("datapoint_chan_backup_size", nil, int64(len(forwarder.dpChan))),
		sfxclient.Gauge("event_chan_backup_size", nil, int64(len(forwarder.eChan))),
		sfxclient.Gauge("datapoint_backup_size", nil, atomic.LoadInt64(&forwarder.stats.totalDatapointsBuffered)),
		sfxclient.Gauge("event_backup_size", nil, atomic.LoadInt64(&forwarder.stats.totalEventsBuffered)),
		sfxclient.Gauge("trace_chan_backup_size", nil, int64(len(forwarder.tChan))),
		sfxclient.Gauge("trace_backup_size", nil, atomic.LoadInt64(&forwarder.stats.totalTracesBuffered)),
	}
}

// DefaultDatapoints does nothing and exists to satisfy the protocol.forwarder interface
func (forwarder *BufferedForwarder) DefaultDatapoints() []*datapoint.Datapoint {
	return []*datapoint.Datapoint{}
}

// Datapoints implements the sfxclient.Collector interface and returns all datapoints about the buffered forwarder
func (forwarder *BufferedForwarder) Datapoints() []*datapoint.Datapoint {
	return append(forwarder.DebugDatapoints(), forwarder.DefaultDatapoints()...)
}

// Pipeline for a BufferedForwarder is the total of all buffers and what is in flight
func (forwarder *BufferedForwarder) Pipeline() int64 {
	return int64(len(forwarder.dpChan)) + int64(len(forwarder.eChan)) + atomic.LoadInt64(&forwarder.stats.datapointsInFlight) + atomic.LoadInt64(&forwarder.stats.eventsInFlight) + int64(len(forwarder.tChan)) + atomic.LoadInt64(&forwarder.stats.tracesInFlight)
}

func (forwarder *BufferedForwarder) blockingDrainUpTo() ([]*datapoint.Datapoint, context.Context) {
	// We block the mutex so we allow one drainer to fully drain until we use another
	forwarder.blockingDrainWaitMutex.Lock()
	defer forwarder.blockingDrainWaitMutex.Unlock()

	var currentToken string
	var datapoints []*datapoint.Datapoint
	if len(forwarder.holdDatapoints) > 0 {
		// use hold, this means we hit a batch that wasn't the same token, next thread takes it
		datapoints = forwarder.holdDatapoints
		currentToken = forwarder.getTokenFromDatapoints(datapoints, currentToken)
		forwarder.holdDatapoints = nil
	} else {
		// Block for at least one point
		select {
		case datapoints = <-forwarder.dpChan:
			if forwarder.useAuthFromRequest {
				// all points in a batch will have the same token
				currentToken = forwarder.getTokenFromDatapoints(datapoints, currentToken)
			}
		case <-forwarder.stopContext.Done():
			return []*datapoint.Datapoint{}, forwarder.stopContext
		}
	}

	datapoints, ctx := forwarder.blockingDrainUpToPost(datapoints, currentToken)
	return datapoints, ctx
}

func (forwarder *BufferedForwarder) blockingDrainUpToPost(datapoints []*datapoint.Datapoint, currentToken string) ([]*datapoint.Datapoint, context.Context) {
Loop:
	for int64(len(datapoints)) < *forwarder.config.MaxDrainSize {
		select {
		case datapoint := <-forwarder.dpChan:
			if forwarder.useAuthFromRequest {
				if forwarder.getTokenFromDatapoints(datapoint, "") != currentToken {
					forwarder.holdDatapoints = datapoint
					break Loop
				}
			}
			datapoints = append(datapoints, datapoint...)
			continue
		default:
			// Nothing left.  Flush this.
			break Loop
		}
	}
	atomic.AddInt64(&forwarder.stats.totalDatapointsBuffered, int64(-len(datapoints)))
	ctx := forwarder.stopContext
	if currentToken != "" {
		ctx = context.WithValue(ctx, sfxclient.TokenHeaderName, currentToken)
	}
	return datapoints, ctx
}

func (forwarder *BufferedForwarder) getTokenFromDatapoints(datapoints []*datapoint.Datapoint, currentToken string) string {
	if tok, ok := datapoints[0].Meta[sfxclient.TokenHeaderName]; ok {
		currentToken = tok.(string) // all points in a batch will have the same token
	}
	return currentToken
}

func (forwarder *BufferedForwarder) blockingDrainEventsUpTo() ([]*event.Event, context.Context) {
	// We block the mutex so we allow one drainer to fully drain until we use another
	forwarder.blockingEventDrainWaitMutex.Lock()
	defer forwarder.blockingEventDrainWaitMutex.Unlock()

	var currentToken string
	var events []*event.Event
	if len(forwarder.holdEvents) > 0 {
		// use hold, this means we hit a batch that wasn't the same token, next thread takes it
		events = forwarder.holdEvents
		currentToken = forwarder.getTokenFromEvents(events, currentToken)
		forwarder.holdEvents = nil
	} else {
		// Block for at least one point
		select {
		case events = <-forwarder.eChan:
			if forwarder.useAuthFromRequest {
				// all points in a batch will have the same token
				currentToken = forwarder.getTokenFromEvents(events, currentToken)
			}
		case <-forwarder.stopContext.Done():
			return []*event.Event{}, forwarder.stopContext
		}
	}

	events, ctx := forwarder.blockingDrainEventsUpToPost(events, currentToken)
	return events, ctx
}

func (forwarder *BufferedForwarder) blockingDrainEventsUpToPost(events []*event.Event, currentToken string) ([]*event.Event, context.Context) {
Loop:
	for int64(len(events)) < *forwarder.config.MaxDrainSize {
		select {
		case event := <-forwarder.eChan:
			if forwarder.useAuthFromRequest {
				if forwarder.getTokenFromEvents(event, "") != currentToken {
					forwarder.holdEvents = event
					break Loop
				}
			}
			events = append(events, event...)
			continue
		default:
			// Nothing left.  Flush this.
			break Loop
		}
	}
	atomic.AddInt64(&forwarder.stats.totalEventsBuffered, int64(-len(events)))
	ctx := forwarder.stopContext
	if currentToken != "" {
		ctx = context.WithValue(ctx, sfxclient.TokenHeaderName, currentToken)
	}
	return events, ctx
}

func (forwarder *BufferedForwarder) getTokenFromEvents(events []*event.Event, currentToken string) string {
	if tok, ok := events[0].Meta[sfxclient.TokenHeaderName]; ok {
		currentToken = tok.(string) // all points in a batch will have the same token
	}
	return currentToken
}

func (forwarder *BufferedForwarder) blockingDrainSpansUpTo() ([]*trace.Span, context.Context) {
	// We block the mutex so we allow one drainer to fully drain until we use another
	forwarder.blockingTraceDrainWaitMutex.Lock()
	defer forwarder.blockingTraceDrainWaitMutex.Unlock()

	var currentToken string
	var spans []*trace.Span
	if len(forwarder.holdSpans) > 0 {
		// use hold, this means we hit a batch that wasn't the same token, next thread takes it
		spans = forwarder.holdSpans
		currentToken = forwarder.getTokenFromSpans(spans, currentToken)
		forwarder.holdSpans = nil
	} else {
		// Block for at least one point
		select {
		case spans = <-forwarder.tChan:
			if forwarder.useAuthFromRequest {
				// all points in a batch will have the same token
				currentToken = forwarder.getTokenFromSpans(spans, currentToken)
			}
		case <-forwarder.stopContext.Done():
			return []*trace.Span{}, forwarder.stopContext
		}
	}

	spans, ctx := forwarder.blockingDrainSpansUpToPost(spans, currentToken)
	return spans, ctx
}

func (forwarder *BufferedForwarder) blockingDrainSpansUpToPost(spans []*trace.Span, currentToken string) ([]*trace.Span, context.Context) {
Loop:
	for int64(len(spans)) < *forwarder.config.MaxDrainSize {
		select {
		case span := <-forwarder.tChan:
			if forwarder.useAuthFromRequest {
				if forwarder.getTokenFromSpans(span, "") != currentToken {
					forwarder.holdSpans = span
					break Loop
				}
			}
			spans = append(spans, span...)
			continue
		default:
			// Nothing left.  Flush this.
			break Loop
		}
	}
	atomic.AddInt64(&forwarder.stats.totalTracesBuffered, int64(-len(spans)))
	ctx := forwarder.stopContext
	if currentToken != "" {
		ctx = context.WithValue(ctx, sfxclient.TokenHeaderName, currentToken)
	}
	return spans, ctx
}

func (forwarder *BufferedForwarder) getTokenFromSpans(spans []*trace.Span, currentToken string) string {
	if tok, ok := spans[0].Meta[sfxclient.TokenHeaderName]; ok {
		currentToken = tok.(string) // all points in a batch will have the same token
	}
	return currentToken
}

// Close stops the threads that are flushing channel points to the next forwarder
func (forwarder *BufferedForwarder) Close() error {
	forwarder.stopFunc()
	forwarder.threadsWaitingToDie.Wait()
	return forwarder.closeSender()
}

func (forwarder *BufferedForwarder) doData(drainIndex int64) {
	defer forwarder.threadsWaitingToDie.Done()
	logger := log.NewContext(forwarder.logger).With("draining_index", drainIndex)
	for forwarder.stopContext.Err() == nil {
		datapoints, ctx := forwarder.blockingDrainUpTo()
		if len(datapoints) == 0 {
			continue
		}
		logDpIfFlag(logger, forwarder.checker, datapoints, "about to send datapoint")
		atomic.AddInt64(&forwarder.stats.datapointsInFlight, int64(len(datapoints)))
		err := forwarder.sendTo.AddDatapoints(ctx, datapoints)
		atomic.AddInt64(&forwarder.stats.datapointsInFlight, -int64(len(datapoints)))
		if err != nil {
			logger.Log(log.Err, err, "error sending datapoints")
		}
		logDpIfFlag(logger, forwarder.checker, datapoints, "Finished sending datapoint")
	}
}

func (forwarder *BufferedForwarder) doEvent(drainIndex int64) {
	defer forwarder.threadsWaitingToDie.Done()
	logger := log.NewContext(forwarder.logger).With("draining_index", drainIndex)
	for forwarder.stopContext.Err() == nil {
		events, ctx := forwarder.blockingDrainEventsUpTo()
		if len(events) == 0 {
			continue
		}
		logEvIfFlag(logger, forwarder.checker, events, "about to send event")
		atomic.AddInt64(&forwarder.stats.eventsInFlight, int64(len(events)))
		err := forwarder.sendTo.AddEvents(ctx, events)
		atomic.AddInt64(&forwarder.stats.eventsInFlight, -int64(len(events)))
		if err != nil {
			logger.Log(log.Err, err, "error sending events")
		}
		logEvIfFlag(logger, forwarder.checker, events, "Finished sending event")
	}
}

func (forwarder *BufferedForwarder) doSpan(drainIndex int64) {
	defer forwarder.threadsWaitingToDie.Done()
	logger := log.NewContext(forwarder.logger).With("draining_index", drainIndex)
	for forwarder.stopContext.Err() == nil {
		traces, ctx := forwarder.blockingDrainSpansUpTo()
		if len(traces) == 0 {
			continue
		}
		atomic.AddInt64(&forwarder.stats.tracesInFlight, int64(len(traces)))
		err := forwarder.sendTo.AddSpans(ctx, traces)
		atomic.AddInt64(&forwarder.stats.tracesInFlight, -int64(len(traces)))
		if err != nil {
			logger.Log(log.Err, err, "error sending traces")
		}
	}
}

func (forwarder *BufferedForwarder) start() {
	forwarder.threadsWaitingToDie.Add(int(*forwarder.config.NumDrainingThreads) * 3)
	for i := int64(0); i < *forwarder.config.NumDrainingThreads; i++ {
		go forwarder.doData(i)
		go forwarder.doEvent(i)
		go forwarder.doSpan(i)
	}
}

type flagChecker interface {
	HasDatapointFlag(*datapoint.Datapoint) bool
	HasEventFlag(*event.Event) bool
}

func logDpIfFlag(l log.Logger, checker flagChecker, dps []*datapoint.Datapoint, msg string) {
	if log.IsDisabled(l) {
		return
	}
	for _, dp := range dps {
		if checker.HasDatapointFlag(dp) {
			l.Log("dp", dp, msg)
		}
	}
}

func logEvIfFlag(l log.Logger, checker flagChecker, ev []*event.Event, msg string) {
	if log.IsDisabled(l) {
		return
	}
	for _, dp := range ev {
		if checker.HasEventFlag(dp) {
			l.Log("ev", dp, msg)
		}
	}
}

// StartupFinished runs the afterStartup method on the forwarder
func (forwarder *BufferedForwarder) StartupFinished() error {
	return forwarder.afterStartup()
}

// NewBufferedForwarder is used only by this package to create a forwarder that buffers its
// datapoint channel
func NewBufferedForwarder(ctx context.Context, config *Config, sendTo Sink, closeIt, afterStartup func() error, logger log.Logger, debugEnpoints func() map[string]http.Handler) *BufferedForwarder {
	config = pointer.FillDefaultFrom(config, DefaultConfig).(*Config)
	logCtx := log.NewContext(logger).With(logkey.Struct, "BufferedForwarder")
	logCtx.Log(logkey.Config, config)
	context, cancel := context.WithCancel(ctx)
	ret := &BufferedForwarder{
		stopFunc:           cancel,
		stopContext:        context,
		dpChan:             make(chan []*datapoint.Datapoint, *config.BufferSize),
		eChan:              make(chan []*event.Event, *config.BufferSize),
		tChan:              make(chan []*trace.Span, *config.BufferSize),
		config:             config,
		sendTo:             sendTo,
		closeSender:        closeIt,
		afterStartup:       afterStartup,
		logger:             logCtx,
		checker:            config.Checker,
		cdim:               config.Cdim,
		identifier:         *config.Name,
		debugEndpoints:     debugEnpoints,
		useAuthFromRequest: *config.UseAuthFromRequest,
	}
	ret.start()
	return ret
}

redpajama_set_name: RedPajamaGithub