package com.cloudera.flume.master;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.avro.AvroRuntimeException;
import org.apache.avro.ipc.AvroRemoteException;
import org.apache.avro.ipc.HttpServer;
import org.apache.avro.specific.SpecificResponder;
import org.junit.Assert;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.cloudera.flume.conf.FlumeConfigData;
import com.cloudera.flume.conf.avro.AvroFlumeConfigData;
import com.cloudera.flume.conf.avro.CommandStatusAvro;
import com.cloudera.flume.conf.avro.FlumeMasterAdminServerAvro;
import com.cloudera.flume.conf.avro.FlumeMasterCommandAvro;
import com.cloudera.flume.conf.avro.FlumeNodeStatusAvro;
import com.cloudera.flume.util.AdminRPC;
import com.cloudera.flume.util.AdminRPCAvro;
public static final Logger LOG = LoggerFactory.getLogger(TestAvroAdminServer.class);
public class MyAvroServer implements FlumeMasterAdminServerAvro {
private HttpServer server;
public void serve()
throws IOException {
LOG.info("Starting dummy server");
SpecificResponder res = new SpecificResponder(
FlumeMasterAdminServerAvro.class, this);
this.server = new HttpServer(res, 56789);
this.server.start();
}
this.server.close();
}
@Override
throws AvroRemoteException {
return new HashMap<CharSequence, FlumeNodeStatusAvro>();
}
@Override
public Map<CharSequence, List<CharSequence>>
getMappings(
CharSequence physicalNode) throws AvroRemoteException {
return new HashMap<CharSequence, List<CharSequence>>();
}
@Override
public boolean isFailure(
long cmdid)
throws AvroRemoteException {
return true;
}
@Override
public boolean isSuccess(
long cmdid)
throws AvroRemoteException {
return false;
}
@Override
public long submit(FlumeMasterCommandAvro command)
throws AvroRemoteException {
return 42;
}
@Override
public Map<CharSequence, AvroFlumeConfigData>
getConfigs()
throws AvroRemoteException {
return new HashMap<CharSequence, AvroFlumeConfigData>();
}
@Override
public boolean hasCmdId(
long cmdid)
throws AvroRemoteException {
return true;
}
@Override
throws AvroRemoteException {
if (cmdid == 1337) {
List<CharSequence> l = new ArrayList<CharSequence>();
l.add("arg1");
l.add("arg2");
CommandStatusAvro csa = new CommandStatusAvro();
csa.cmdId = cmdid;
csa.state = CommandStatus.State.SUCCEEDED.toString();
csa.message = "message";
csa.cmd = new FlumeMasterCommandAvro();
csa.cmd.command = "cmd";
csa.cmd.arguments = l;
return csa;
}
return null;
}
}
@Test
MyAvroServer server = new MyAvroServer();
server.serve();
AdminRPC client = new AdminRPCAvro("localhost", 56789);
LOG.info("Connected to test master");
long submit = client.submit(new Command(""));
Assert.assertEquals("Expected response was 42, got " + submit, submit, 42);
boolean succ = client.isSuccess(42);
Assert
.assertEquals("Expected response was false, got " + succ, succ, false);
boolean fail = client.isFailure(42);
Assert.assertEquals("Expected response was true, got " + fail, fail, true);
Map<String, FlumeConfigData> cfgs = client.getConfigs();
Assert.assertEquals("Expected response was 0, got " + cfgs.size(), cfgs
.size(), 0);
CommandStatus cs = client.getCommandStatus(1337);
Assert.assertEquals(1337, cs.getCmdID());
Assert.assertEquals("message", cs.getMessage());
Command cmd = cs.getCommand();
Assert.assertEquals("cmd", cmd.getCommand());
Assert.assertEquals("arg1", cmd.getArgs()[0]);
Assert.assertEquals("arg2", cmd.getArgs()[1]);
server.stop();
}
MyAvroServer server = new MyAvroServer();
server.serve();
AdminRPC client = new AdminRPCAvro("localhost", 56789);
LOG.info("Connected to test master");
try {
CommandStatus bad = client.getCommandStatus(1234);
} catch (AvroRuntimeException are) {
return;
} finally {
server.stop();
}
Assert.fail("should have thrown exception");
}
@Test
MyAvroServer server = new MyAvroServer();
for (int i = 0; i < 50; i++) {
LOG.info("open close " + i);
server.serve();
server.stop();
}
}
}