Skip to content

Commit

Permalink
FAB-8074 Misc fixes see description
Browse files Browse the repository at this point in the history 
Change-Id: I7bfcb0a312a6000def95376cfc3cb8c821086265
Signed-off-by: rickr <cr22rc@gmail.com>
  • Loading branch information
@cr22rc
cr22rc committed Feb 5, 2018
1 parent e478cb9 commit 2adcdbc
Show file tree
Hide file tree
Showing 7 changed files with 71 additions and 22 deletions.
10 changes: 9 additions & 1 deletion src/main/java/org/hyperledger/fabric/sdk/Channel.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2300,7 +2300,15 @@ private Collection<ProposalResponse> sendProposal(TransactionRequest proposalReq
checkPeers(peers);

if (null == proposalRequest) {
throw new InvalidArgumentException("sendProposal queryProposalRequest is null");
throw new InvalidArgumentException("The proposalRequest is null");
}

if (Utils.isNullOrEmpty(proposalRequest.getFcn())) {
throw new InvalidArgumentException("The proposalRequest's fcn is null or empty.");
}

if (proposalRequest.getChaincodeID() == null) {
throw new InvalidArgumentException("The proposalRequest's chaincode ID is null");
}

proposalRequest.setSubmitted();
Expand Down
5 changes: 4 additions & 1 deletion src/main/java/org/hyperledger/fabric/sdk/HFClient.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -278,7 +278,7 @@ public Channel deSerializeChannel(byte[] channelBytes)
/**
* newPeer create a new peer
*
* @param name
* @param name name of peer.
* @param grpcURL to the peer's location
* @param properties <p>
* Supported properties
Expand All @@ -302,6 +302,9 @@ public Channel deSerializeChannel(byte[] channelBytes)
* hostname verifications during TLS handshake.
* </li>
* <li>
* peerEventRegistrationWaitTime - Time in milliseconds to wait for peer eventing service registration.
* </li>
* <li>
* grpc.NettyChannelBuilderOption.&lt;methodName&gt; where methodName is any method on
* grpc ManagedChannelBuilder. If more than one argument to the method is needed then the
* parameters need to be supplied in an array of Objects.
Expand Down
7 changes: 6 additions & 1 deletion src/main/java/org/hyperledger/fabric/sdk/OrdererClient.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -141,7 +141,12 @@ Ab.BroadcastResponse sendTransaction(Common.Envelope envelope) throws Exception
public void onNext(Ab.BroadcastResponse resp) {
// logger.info("Got Broadcast response: " + resp);
logger.debug("resp status value: " + resp.getStatusValue() + ", resp: " + resp.getStatus());
ret[0] = resp;
if (resp.getStatus() == Common.Status.SUCCESS) {
ret[0] = resp;
} else {
throwable[0] = new TransactionException(format("Channel %s orderer %s status returned failure code %d (%s) during order registration",
channelName, name, resp.getStatusValue(), resp.getStatus().name()));
}
finishLatch.countDown();

}
Expand Down
45 changes: 26 additions & 19 deletions src/main/java/org/hyperledger/fabric/sdk/PeerEventServiceClient.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import io.grpc.stub.StreamObserver;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.hyperledger.fabric.protos.common.Common;
import org.hyperledger.fabric.protos.common.Common.Envelope;
import org.hyperledger.fabric.protos.orderer.Ab;
import org.hyperledger.fabric.protos.orderer.Ab.SeekInfo;
Expand All @@ -45,19 +46,18 @@
import static org.hyperledger.fabric.protos.peer.PeerEvents.DeliverResponse.TypeCase.STATUS;
import static org.hyperledger.fabric.sdk.transaction.ProtoUtils.createSeekInfoEnvelope;


/**
* Sample client code that makes gRPC calls to the server.
*/
class PeerEventServiceClient {
private static final Config config = Config.getConfig();
private static final long ORDERER_WAIT_TIME = config.getOrdererWaitTime();
private static final long PEER_EVENT_REGISTRATION_WAIT_TIME = config.getPeerEventRegistrationWaitTime();
private static final Log logger = LogFactory.getLog(PeerEventServiceClient.class);
private final String channelName;
private final ManagedChannelBuilder channelBuilder;
private final String name;
private final String url;
private final long ordererWaitTimeMilliSecs;
private final long peerEventRegistrationWaitTimeMilliSecs;
private final PeerOptions peerOptions;
private final boolean filterBlock;
Properties properties = new Properties();
Expand All @@ -70,7 +70,7 @@ class PeerEventServiceClient {
private transient Peer peer;

/**
* Construct client for accessing Orderer server using the existing managedChannel.
* Construct client for accessing Peer eventing service using the existing managedChannel.
*/
PeerEventServiceClient(Peer peer, ManagedChannelBuilder<?> channelBuilder, Properties properties, PeerOptions peerOptions) {

Expand All @@ -86,22 +86,22 @@ class PeerEventServiceClient {

if (null == properties) {

ordererWaitTimeMilliSecs = ORDERER_WAIT_TIME;
peerEventRegistrationWaitTimeMilliSecs = PEER_EVENT_REGISTRATION_WAIT_TIME;

} else {
this.properties = properties;

String ordererWaitTimeMilliSecsString = properties.getProperty("ordererWaitTimeMilliSecs", Long.toString(ORDERER_WAIT_TIME));
String peerEventRegistrationWaitTime = properties.getProperty("peerEventRegistrationWaitTime", Long.toString(PEER_EVENT_REGISTRATION_WAIT_TIME));

long tempOrdererWaitTimeMilliSecs = ORDERER_WAIT_TIME;
long tempPeerWaitTimeMilliSecs = PEER_EVENT_REGISTRATION_WAIT_TIME;

try {
tempOrdererWaitTimeMilliSecs = Long.parseLong(ordererWaitTimeMilliSecsString);
tempPeerWaitTimeMilliSecs = Long.parseLong(peerEventRegistrationWaitTime);
} catch (NumberFormatException e) {
logger.warn(format("Orderer %s wait time %s not parsable.", name, ordererWaitTimeMilliSecsString), e);
logger.warn(format("Peer event service registration %s wait time %s not parsable.", name, peerEventRegistrationWaitTime), e);
}

ordererWaitTimeMilliSecs = tempOrdererWaitTimeMilliSecs;
peerEventRegistrationWaitTimeMilliSecs = tempPeerWaitTimeMilliSecs;
}

}
Expand Down Expand Up @@ -202,21 +202,30 @@ public void onNext(DeliverResponse resp) {
done = true;
logger.debug(format("DeliverResponse channel %s peer %s setting done.",
channelName, peer.getName()));
retList.add(0, resp);

finishLatch.countDown();
if (resp.getStatus() == Common.Status.SUCCESS) {
retList.add(0, resp);
} else {

throwableList.add(new TransactionException(format("Channel %s peer %s Status returned failure code %d (%s) during peer service event registration",
channelName, peer.getName(), resp.getStatusValue(), resp.getStatus().name())));
}

} else if (typeCase == FILTERED_BLOCK || typeCase == BLOCK) {
logger.trace(format("Channel %s peer %s got event block hex hashcode: %016x, block number: %d",
channelName, peer.getName(), resp.getBlock().hashCode(), resp.getBlock().getHeader().getNumber()));
retList.add(resp);
finishLatch.countDown();

channelEventQue.addBEvent(new BlockEvent(peer, resp));
} else {
logger.error(format("Channel %s peer %s got event block with unknown type: %s, %d",
channelName, peer.getName(), typeCase.name(), typeCase.getNumber())
);
channelName, peer.getName(), typeCase.name(), typeCase.getNumber()));

throwableList.add(new TransactionException(format("Channel %s peer %s Status got unknown type %s, %d",
channelName, peer.getName(), typeCase.name(), typeCase.getNumber())));

}
finishLatch.countDown();

}

Expand All @@ -229,7 +238,6 @@ public void onError(Throwable t) {
if (!shutdown) {
logger.error(format("Received error on channel %s, peer %s, url %s, %s",
channelName, name, url, t.getMessage()), t);

done = true;
throwableList.add(t);
finishLatch.countDown();
Expand Down Expand Up @@ -259,10 +267,9 @@ public void onCompleted() {
//nso.onCompleted();

try {
// if (!finishLatch.await(ordererWaitTimeMilliSecs, TimeUnit.MILLISECONDS)) {
if (!finishLatch.await(9999999, TimeUnit.MILLISECONDS)) {
if (!finishLatch.await(peerEventRegistrationWaitTimeMilliSecs, TimeUnit.MILLISECONDS)) {
TransactionException ex = new TransactionException(format(
"Channel %s connect time exceeded for peer eventing service %s, timed out at %d ms.", channelName, name, ordererWaitTimeMilliSecs));
"Channel %s connect time exceeded for peer eventing service %s, timed out at %d ms.", channelName, name, peerEventRegistrationWaitTimeMilliSecs));
logger.error(ex.getMessage(), ex);
throw ex;
}
Expand Down
11 changes: 11 additions & 0 deletions src/main/java/org/hyperledger/fabric/sdk/helper/Config.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ public class Config {
public static final String TRANSACTION_CLEANUP_UP_TIMEOUT_WAIT_TIME = "org.hyperledger.fabric.sdk.client.transaction_cleanup_up_timeout_wait_time";
public static final String ORDERER_RETRY_WAIT_TIME = "org.hyperledger.fabric.sdk.orderer_retry.wait_time";
public static final String ORDERER_WAIT_TIME = "org.hyperledger.fabric.sdk.orderer.ordererWaitTimeMilliSecs";
public static final String PEER_EVENT_REGISTRATION_WAIT_TIME = "org.hyperledger.fabric.sdk.peer.eventRegistration.wait_time";
public static final String EVENTHUB_CONNECTION_WAIT_TIME = "org.hyperledger.fabric.sdk.eventhub_connection.wait_time";
public static final String GENESISBLOCK_WAIT_TIME = "org.hyperledger.fabric.sdk.channel.genesisblock_wait_time";
/**
Expand Down Expand Up @@ -105,6 +106,7 @@ private Config() {
defaultProperty(CHANNEL_CONFIG_WAIT_TIME, "15000");
defaultProperty(ORDERER_RETRY_WAIT_TIME, "200");
defaultProperty(ORDERER_WAIT_TIME, "10000");
defaultProperty(PEER_EVENT_REGISTRATION_WAIT_TIME, "2000");
defaultProperty(EVENTHUB_CONNECTION_WAIT_TIME, "1000");
defaultProperty(GENESISBLOCK_WAIT_TIME, "5000");
/**
Expand Down Expand Up @@ -348,6 +350,15 @@ public long getOrdererWaitTime() {
return Long.parseLong(getProperty(ORDERER_WAIT_TIME));
}

/**
* getPeerEventRegistrationWaitTime
*
* @return time in milliseconds to wait for peer eventing service to wait for event registration
*/
public long getPeerEventRegistrationWaitTime() {
return Long.parseLong(getProperty(PEER_EVENT_REGISTRATION_WAIT_TIME));
}

public long getEventHubConnectionWaitTime() {
return Long.parseLong(getProperty(EVENTHUB_CONNECTION_WAIT_TIME));
}
Expand Down
9 changes: 9 additions & 0 deletions src/test/java/org/hyperledger/fabric/sdkintegration/End2endAndBackAgainIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -225,11 +225,17 @@ public void setup() {
SampleOrg sampleOrg = testConfig.getIntegrationTestsSampleOrg("peerOrg1");
Channel fooChannel = reconstructChannel(FOO_CHANNEL_NAME, client, sampleOrg);
runChannel(client, fooChannel, sampleOrg, 0);
assertFalse(fooChannel.isShutdown());
assertTrue(fooChannel.isInitialized());
fooChannel.shutdown(true); //clean up resources no longer needed.
assertTrue(fooChannel.isShutdown());
out("\n");

sampleOrg = testConfig.getIntegrationTestsSampleOrg("peerOrg2");
Channel barChannel = reconstructChannel(BAR_CHANNEL_NAME, client, sampleOrg);
runChannel(client, barChannel, sampleOrg, 100); //run a newly constructed foo channel with different b value!
assertFalse(barChannel.isShutdown());
assertTrue(barChannel.isInitialized());

if (!testConfig.isRunningAgainstFabric10()) { //Peer eventing service support started with v1.1

Expand Down Expand Up @@ -696,6 +702,9 @@ private Channel reconstructChannel(String name, HFClient client, SampleOrg sampl

}

assertTrue(newChannel.isInitialized());
assertFalse(newChannel.isShutdown());

out("Finished reconstructing channel %s.", name);

return newChannel;
Expand Down
6 changes: 6 additions & 0 deletions src/test/java/org/hyperledger/fabric/sdkintegration/End2endIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -244,22 +244,28 @@ sampleOrgDomainName, format("/users/Admin@%s/msp/keystore", sampleOrgDomainName)
Channel fooChannel = constructChannel(FOO_CHANNEL_NAME, client, sampleOrg);
runChannel(client, fooChannel, true, sampleOrg, 0);

assertFalse(fooChannel.isShutdown());
fooChannel.shutdown(true); // Force foo channel to shutdown clean up resources.
assertTrue(fooChannel.isShutdown());

assertNull(client.getChannel(FOO_CHANNEL_NAME));
out("\n");

sampleOrg = testConfig.getIntegrationTestsSampleOrg("peerOrg2");
Channel barChannel = constructChannel(BAR_CHANNEL_NAME, client, sampleOrg);
assertTrue(barChannel.isInitialized());
/**
* sampleStore.saveChannel uses {@link Channel#serializeChannel()}
*/
sampleStore.saveChannel(barChannel);
assertFalse(barChannel.isShutdown());
runChannel(client, barChannel, true, sampleOrg, 100); //run a newly constructed bar channel with different b value!
//let bar channel just shutdown so we have both scenarios.

out("\nTraverse the blocks for chain %s ", barChannel.getName());
blockWalker(client, barChannel);
assertFalse(barChannel.isShutdown());
assertTrue(barChannel.isInitialized());
out("That's all folks!");

} catch (Exception e) {
Expand Down

0 comments on commit 2adcdbc

Please sign in to comment.